feat: Social page — aggregated post queues + X/video history (#345)

* feat(api): x/video post history endpoints

Approved or rejected drafts vanished from both queues permanently --
the listers exclude terminal statuses and no history surface existed,
so a posted tweet or video was only findable in the raw task list.
GET /x/posts/history and GET /video/posts/history (CEO-gated, bounded)
return acted-on drafts newest-first with the posted platform ids and
reject reasons from the draft markers. Route tests assert by identity,
not emptiness: approve/reject commits the whole session, so prior
tests' rows legitimately persist in the shared test DB.

* feat(panel): Social page aggregating post queues and history

New dashboard page composing the X and video post queues with one
unified history section beneath them -- both platforms interleaved
newest-first, kind and outcome badges, posted X ids linking to the
live tweet, reject reasons shown. The command center's two full queue
cards become a compact pending-counts card linking to the page, so the
queues have one home instead of duplicated surfaces.

---------

Co-authored-by: Renn F <rennf93@users.noreply.github.com>

tests/integration/test_video_routes.py

@@ -44,6 +44,7 @@ SLUG = "roboco-video-route-test"
 SYSTEM_UUID = _foundation.AGENTS["system"].uuid
 UX_DEV_1_UUID = _foundation.AGENTS["ux-dev-1"].uuid
 UX_DEV_2_UUID = _foundation.AGENTS["ux-dev-2"].uuid
+HISTORY_LIMIT = 2
 
 
 async def _seed(session: AsyncSession) -> None:
@@ -438,6 +439,99 @@ async def test_approve_with_credentials_posts_via_the_real_poster_wiring(
         await db_session.commit()
 
 
+@pytest.mark.asyncio
+async def test_history_returns_posted_and_rejected_newest_first(
+    db_session: AsyncSession, ceo_client: AsyncClient
+) -> None:
+    rejected = await _seed_draft(db_session, platforms=["x"])
+    with _LOCKED[0], _LOCKED[1]:
+        await ceo_client.post(
+            f"/api/video/posts/{rejected.id}/reject",
+            json={"reason": "wrong occasion"},
+        )
+    posted = await _seed_draft(db_session, platforms=["x"])
+    creds_svc = get_x_credentials_service(db_session)
+    await creds_svc.set_credentials(
+        api_key="ak", api_secret="as", access_token="at", access_token_secret="ats"
+    )
+    try:
+        with (
+            _LOCKED[0],
+            _LOCKED[1],
+            patch.object(
+                LiveXVideoPoster,
+                "post_video",
+                AsyncMock(
+                    return_value=XVideoPostResult(
+                        posted=True, video_id="xid42", detail="posted"
+                    )
+                ),
+            ),
+        ):
+            await ceo_client.post(f"/api/video/posts/{posted.id}/approve", json={})
+
+        resp = await ceo_client.get("/api/video/posts/history")
+        assert resp.status_code == HTTPStatus.OK
+        body = resp.json()
+        ids = [row["task_id"] for row in body]
+        assert str(posted.id) in ids
+        assert str(rejected.id) in ids
+        assert ids.index(str(posted.id)) < ids.index(str(rejected.id))
+        posted_row = next(row for row in body if row["task_id"] == str(posted.id))
+        assert posted_row["status"] == "completed"
+        assert posted_row["posted"] == {"x": "xid42"}
+        rejected_row = next(row for row in body if row["task_id"] == str(rejected.id))
+        assert rejected_row["status"] == "cancelled"
+        assert rejected_row["reject_reason"] == "wrong occasion"
+    finally:
+        await creds_svc.set_credentials(
+            api_key="", api_secret="", access_token="", access_token_secret=""
+        )
+        await db_session.commit()
+
+
+@pytest.mark.asyncio
+async def test_history_excludes_open_drafts(
+    db_session: AsyncSession, ceo_client: AsyncClient
+) -> None:
+    """Every approve/reject route in this file commits durably (the route
+    always calls db.commit()), so other tests' posted/rejected rows persist
+    in this shared-DB test session — history is never provably empty. Assert
+    identity instead: THIS still-open draft must not appear."""
+    open_task = await _seed_draft(db_session)
+    resp = await ceo_client.get("/api/video/posts/history")
+    assert resp.status_code == HTTPStatus.OK
+    ids = [row["task_id"] for row in resp.json()]
+    assert str(open_task.id) not in ids
+
+
+@pytest.mark.asyncio
+async def test_history_respects_limit(
+    db_session: AsyncSession, ceo_client: AsyncClient
+) -> None:
+    for _ in range(3):
+        t = await _seed_draft(db_session)
+        with _LOCKED[0], _LOCKED[1]:
+            await ceo_client.post(
+                f"/api/video/posts/{t.id}/reject", json={"reason": "not relevant"}
+            )
+    resp = await ceo_client.get(
+        "/api/video/posts/history", params={"limit": HISTORY_LIMIT}
+    )
+    assert resp.status_code == HTTPStatus.OK
+    assert len(resp.json()) == HISTORY_LIMIT
+
+
+@pytest.mark.asyncio
+async def test_history_non_ceo_is_forbidden(db_session: AsyncSession) -> None:
+    app = _build_app(db_session, AgentRole.DEVELOPER, uuid4())
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as client:
+        resp = await client.get("/api/video/posts/history")
+    assert resp.status_code == HTTPStatus.FORBIDDEN
+    app.dependency_overrides.clear()
+
+
 @pytest.mark.asyncio
 async def test_approve_edited_x_caption_over_limit_is_422(
     db_session: AsyncSession, ceo_client: AsyncClient

tests/integration/test_x_routes.py

@@ -22,6 +22,8 @@ from roboco.services.task import X_POST_SOURCE
 from roboco.services.x_client import XClient, XMention, XPostResult
 from roboco.services.x_post_service import XPostService
 
+HISTORY_LIMIT = 2
+
 if TYPE_CHECKING:
     from collections.abc import AsyncIterator
 
@@ -193,6 +195,79 @@ async def test_reject_cancels_and_records_reason(
     assert refreshed.status == TaskStatus.CANCELLED
 
 
+@pytest.mark.asyncio
+async def test_history_returns_posted_and_rejected_newest_first(
+    db_session: AsyncSession, ceo_client: AsyncClient
+) -> None:
+    rejected = await _seed_draft(db_session)
+    await ceo_client.post(
+        f"/api/x/posts/{rejected.id}/reject", json={"reason": "off-brand tone"}
+    )
+    posted = await _seed_draft(db_session)
+    with (
+        patch(
+            "roboco.services.x_post_service.build_x_client",
+            return_value=_StubClient(),
+        ),
+        patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
+        patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
+    ):
+        await ceo_client.post(f"/api/x/posts/{posted.id}/approve", json={})
+
+    resp = await ceo_client.get("/api/x/posts/history")
+    assert resp.status_code == HTTPStatus.OK
+    body = resp.json()
+    ids = [row["task_id"] for row in body]
+    assert str(posted.id) in ids
+    assert str(rejected.id) in ids
+    assert ids.index(str(posted.id)) < ids.index(str(rejected.id))
+    posted_row = next(row for row in body if row["task_id"] == str(posted.id))
+    assert posted_row["status"] == "completed"
+    assert posted_row["tweet_id"] == "42"
+    rejected_row = next(row for row in body if row["task_id"] == str(rejected.id))
+    assert rejected_row["status"] == "cancelled"
+    assert rejected_row["reject_reason"] == "off-brand tone"
+
+
+@pytest.mark.asyncio
+async def test_history_excludes_open_drafts(
+    db_session: AsyncSession, ceo_client: AsyncClient
+) -> None:
+    """Every approve/reject route in this file commits durably (the route
+    always calls db.commit()), so other tests' posted/rejected rows persist
+    in this shared-DB test session — history is never provably empty. Assert
+    identity instead: THIS still-open draft must not appear."""
+    open_task = await _seed_draft(db_session)
+    resp = await ceo_client.get("/api/x/posts/history")
+    assert resp.status_code == HTTPStatus.OK
+    ids = [row["task_id"] for row in resp.json()]
+    assert str(open_task.id) not in ids
+
+
+@pytest.mark.asyncio
+async def test_history_respects_limit(
+    db_session: AsyncSession, ceo_client: AsyncClient
+) -> None:
+    for _ in range(3):
+        t = await _seed_draft(db_session)
+        await ceo_client.post(
+            f"/api/x/posts/{t.id}/reject", json={"reason": "not relevant"}
+        )
+    resp = await ceo_client.get("/api/x/posts/history", params={"limit": HISTORY_LIMIT})
+    assert resp.status_code == HTTPStatus.OK
+    assert len(resp.json()) == HISTORY_LIMIT
+
+
+@pytest.mark.asyncio
+async def test_history_non_ceo_is_forbidden(db_session: AsyncSession) -> None:
+    app = _build_app(db_session, AgentRole.DEVELOPER, uuid4())
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as client:
+        resp = await client.get("/api/x/posts/history")
+    assert resp.status_code == HTTPStatus.FORBIDDEN
+    app.dependency_overrides.clear()
+
+
 @pytest.mark.asyncio
 async def test_credentials_default_is_unset(ceo_client: AsyncClient) -> None:
     resp = await ceo_client.get("/api/x/credentials")

tests/unit/services/test_video_post_service.py

@@ -577,6 +577,98 @@ async def test_list_held_video_posts_excludes_terminal(
     assert rejected_task.id not in ids
 
 
+@pytest.mark.asyncio
+async def test_list_video_post_history_excludes_open_drafts(
+    db_session: AsyncSession, _test_database_url: str
+) -> None:
+    """approve() commits the whole session, so open_task's still-uncommitted
+    seed insert becomes durable too (same class of leak documented on
+    test_approve_partial_failure_keeps_task_open_and_persists_the_success) —
+    clean it up explicitly so it doesn't pollute list_open_video_posts()/
+    list_open_video_post_drafts() assertions elsewhere in the suite."""
+    open_task = await _seed_video_post(db_session)
+    open_task_id = _id(open_task)
+    open_project_id = open_task.project_id
+    posted_task = await _seed_video_post(db_session, platforms=["x"])
+    svc = _svc(db_session, x_poster=_StubXPoster(), tiktok_poster=_StubTikTokPoster())
+    try:
+        with _LOCKED[0], _LOCKED[1]:
+            await svc.approve(_id(posted_task))
+        history = await svc.list_video_post_history()
+        ids = {t.id for t in history}
+        assert posted_task.id in ids
+        assert open_task.id not in ids
+    finally:
+        cleanup, cleanup_engine = await _fresh_session(_test_database_url)
+        try:
+            await cleanup.execute(delete(TaskTable).where(TaskTable.id == open_task_id))
+            await cleanup.execute(
+                delete(ProjectTable).where(ProjectTable.id == open_project_id)
+            )
+            await cleanup.commit()
+        finally:
+            await _dispose(cleanup, cleanup_engine)
+
+
+@pytest.mark.asyncio
+async def test_list_video_post_history_newest_acted_first(
+    db_session: AsyncSession,
+) -> None:
+    rejected_task = await _seed_video_post(db_session)
+    svc = _svc(db_session, x_poster=_StubXPoster(), tiktok_poster=_StubTikTokPoster())
+    with _LOCKED[0], _LOCKED[1]:
+        await svc.reject(_id(rejected_task), "wrong occasion")
+    posted_task = await _seed_video_post(db_session, platforms=["x"])
+    with _LOCKED[0], _LOCKED[1]:
+        await svc.approve(_id(posted_task))
+    history = await svc.list_video_post_history()
+    ids = [t.id for t in history]
+    assert ids.index(posted_task.id) < ids.index(rejected_task.id)
+
+
+@pytest.mark.asyncio
+async def test_list_video_post_history_includes_marker_fields(
+    db_session: AsyncSession,
+) -> None:
+    posted_task = await _seed_video_post(db_session, platforms=["x"])
+    svc = _svc(
+        db_session,
+        x_poster=_StubXPoster(video_id="xid9"),
+        tiktok_poster=_StubTikTokPoster(),
+    )
+    with _LOCKED[0], _LOCKED[1]:
+        await svc.approve(_id(posted_task))
+    rejected_task = await _seed_video_post(db_session)
+    with _LOCKED[0], _LOCKED[1]:
+        await svc.reject(_id(rejected_task), "off-brand")
+
+    history = await svc.list_video_post_history()
+    by_id = {t.id: t for t in history}
+    posted_draft = markers.get_video_draft(by_id[posted_task.id])
+    assert posted_draft is not None
+    assert posted_draft["x_posted_id"] == "xid9"
+    assert markers.get_video_reject_reason(by_id[rejected_task.id]) == "off-brand"
+
+
+@pytest.mark.asyncio
+async def test_list_video_post_history_respects_limit(
+    db_session: AsyncSession,
+) -> None:
+    svc = _svc(db_session, x_poster=_StubXPoster(), tiktok_poster=_StubTikTokPoster())
+    tasks = []
+    for _ in range(3):
+        t = await _seed_video_post(db_session)
+        with _LOCKED[0], _LOCKED[1]:
+            await svc.reject(_id(t), "not relevant")
+        tasks.append(t)
+    history = await svc.list_video_post_history(limit=2)
+    assert len(history) == TWO
+    ids = {t.id for t in history}
+    assert tasks[2].id in ids
+    assert tasks[1].id in ids
+    assert tasks[0].id not in ids
+
+
 @pytest.mark.asyncio
 async def test_approve_commits_before_releasing_the_lock(
     db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch

tests/unit/services/test_x_post_service.py

@@ -42,6 +42,7 @@ if TYPE_CHECKING:
 SYSTEM_UUID = _foundation.AGENTS["system"].uuid
 SECRETARY_UUID = _foundation.AGENTS["secretary-1"].uuid
 ONE = 1
+TWO = 2
 
 
 class _StubClient(XClient):
@@ -364,6 +365,74 @@ async def test_reject_completed_raises(db_session: AsyncSession) -> None:
         await _svc(db_session).reject(_id(task), "nope")
 
 
+@pytest.mark.asyncio
+async def test_list_post_history_excludes_open_drafts(
+    db_session: AsyncSession,
+) -> None:
+    open_task = await _seed_draft(db_session)
+    rejected_task = await _seed_draft(db_session, source=X_REPLY_SOURCE)
+    await _svc(db_session).reject(_id(rejected_task), "not relevant")
+    history = await _svc(db_session).list_post_history()
+    ids = {t.id for t in history}
+    assert rejected_task.id in ids
+    assert open_task.id not in ids
+
+
+@pytest.mark.asyncio
+async def test_list_post_history_newest_acted_first(
+    db_session: AsyncSession,
+) -> None:
+    rejected_task = await _seed_draft(db_session, source=X_REPLY_SOURCE)
+    await _svc(db_session).reject(_id(rejected_task), "not relevant")
+    posted_task = await _seed_draft(db_session)
+    client = _StubClient()
+    with (
+        patch("roboco.services.x_post_service.build_x_client", return_value=client),
+        patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
+        patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
+    ):
+        await _svc(db_session).approve(_id(posted_task))
+    history = await _svc(db_session).list_post_history()
+    ids = [t.id for t in history]
+    assert ids.index(posted_task.id) < ids.index(rejected_task.id)
+
+
+@pytest.mark.asyncio
+async def test_list_post_history_includes_marker_fields(
+    db_session: AsyncSession,
+) -> None:
+    posted_task = await _seed_draft(db_session)
+    client = _StubClient(tweet_id="777")
+    with (
+        patch("roboco.services.x_post_service.build_x_client", return_value=client),
+        patch.object(XPostService, "_acquire_lock", AsyncMock(return_value="tok")),
+        patch.object(XPostService, "_release_lock", AsyncMock(return_value=None)),
+    ):
+        await _svc(db_session).approve(_id(posted_task))
+    rejected_task = await _seed_draft(db_session, source=X_REPLY_SOURCE)
+    await _svc(db_session).reject(_id(rejected_task), "off-brand tone")
+
+    history = await _svc(db_session).list_post_history()
+    by_id = {t.id: t for t in history}
+    assert markers.get_x_posted_tweet_id(by_id[posted_task.id]) == "777"
+    assert markers.get_x_reject_reason(by_id[rejected_task.id]) == "off-brand tone"
+
+
+@pytest.mark.asyncio
+async def test_list_post_history_respects_limit(db_session: AsyncSession) -> None:
+    tasks = []
+    for _ in range(3):
+        t = await _seed_draft(db_session, source=X_REPLY_SOURCE)
+        await _svc(db_session).reject(_id(t), "not relevant")
+        tasks.append(t)
+    history = await _svc(db_session).list_post_history(limit=2)
+    assert len(history) == TWO
+    ids = {t.id for t in history}
+    assert tasks[2].id in ids
+    assert tasks[1].id in ids
+    assert tasks[0].id not in ids
+
+
 @pytest.mark.asyncio
 async def test_approve_does_not_flush_edited_body_before_lock(
     db_session: AsyncSession,