From 4585a248ce39a5eac30a2edff5d25679f4d0b9ed Mon Sep 17 00:00:00 2001 From: Renzo F <45401804+rennf93@users.noreply.github.com> Date: Wed, 22 Jul 2026 20:30:01 +0200 Subject: [PATCH] =?UTF-8?q?feat(x):=20redraft=20loop=20on=20CEO=20reject?= =?UTF-8?q?=20=E2=80=94=20feedback=20re-enters=20the=20draft=20flow=20(#64?= =?UTF-8?q?8)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(x): redraft loop on CEO reject — feedback re-enters the draft flow A rejected X draft's reason used to die with the cancel. reject() with a non-blank reason now schedules a redraft after its commit (defer_after_commit; fresh session; never blocks or fails the HTTP response): XEngine.redraft_from_rejection re-drafts the same source kind via the local model with the reason and rejected body folded in as revision guidance, originating ONE fresh held draft — mirroring the video pipeline's reauthor_from_rejection. Local-model failure or empty output originates nothing (no degraded copies); markers carry forward whole so a redrafted reply/spotlight stays fully functional downstream; bodies ride the same 280 clamp; the open-posts cap holds. Hardened per adversarial review: reject() is now idempotent on an already-CANCELLED target at both check sites (mirroring approve's already_rejected guard — a replayed reject schedules nothing), and the dedup check+originate runs under a non-blocking identity-keyed Redis lock (SET NX + compare-and-del, matching the approve/reject mutex style) so racing rejects can't stack duplicate drafts — lock held or Redis down skips the redraft, which is always safe. Tests pin the fresh-session contract by session identity, the replay no-op, the lock-skip, and clean up their own committed rows. * fix(tests): runtime UUID import + typed task-id coercion in x cleanup helper CI's quality gate runs mypy over tests/ (the local pass covered only roboco/): the _delete_tasks calls handed ORM-typed ids where uuid.UUID was expected. Coercing at the call sites then exposed that UUID was imported under TYPE_CHECKING only — a runtime NameError. Import moved to runtime; both call sites coerce explicitly. --------- Co-authored-by: Renn F --- roboco/services/x_engine.py | 306 +++++++++++++++++++++ roboco/services/x_post_service.py | 122 ++++++-- tests/unit/services/test_x_engine.py | 265 +++++++++++++++++- tests/unit/services/test_x_post_service.py | 299 +++++++++++++++++++- 4 files changed, 972 insertions(+), 20 deletions(-) diff --git a/roboco/services/x_engine.py b/roboco/services/x_engine.py index f569e277..6850d143 100644 --- a/roboco/services/x_engine.py +++ b/roboco/services/x_engine.py @@ -31,6 +31,7 @@ import re from datetime import UTC, datetime, timedelta from pathlib import Path from typing import TYPE_CHECKING, Any, cast +from uuid import uuid4 import httpx import redis.asyncio as redis @@ -57,6 +58,7 @@ from roboco.services.task import ( X_FEATURE_SOURCE, X_POST_SOURCE, X_REPLY_SOURCE, + X_SOURCES, TaskCreateRequest, get_task_service, ) @@ -83,6 +85,30 @@ _MIN_MENTION_CHARS = 3 # Redis TTL guard. _BRAND_VOICE_NUDGE_KEY = "x_brand_voice_nudge_sent" +# redraft-dedup lock (see _redraft_from_rejection): closes the check-then-act +# TOCTOU on _redraft_already_open — two deferred closures racing for the SAME +# identity would otherwise both pass the dedup check (each sees nothing +# committed yet from the other under READ COMMITTED) and both originate. +# Plain SET NX + Lua compare-and-del, mirroring XPostService._acquire_lock / +# _release_lock (same style, a different key namespace) rather than a shared +# helper — this codebase duplicates this small pattern per service +# (release_proposal.py, prompter.py, x_post_service.py all carry their own). +_REDRAFT_LOCK_PREFIX = "roboco:x_redraft:" +_REDRAFT_LOCK_TTL_SECONDS = 60 # the check+insert completes in ms; crash backstop +_REDRAFT_RELEASE_SCRIPT = """ +if redis.call("get", KEYS[1]) == ARGV[1] then + return redis.call("del", KEYS[1]) +else + return 0 +end +""" + + +class _RedraftLockUnavailable(Exception): + """Redis is unreachable for the redraft-dedup lock — distinct from "the + lock is held" (both end in the same skip-the-redraft outcome), mirroring + XPostService's _LockUnavailable.""" + # Ported from agents/prompts/identities/head-marketing.md's VOICE GUIDE (the # reasoning-backed voice HoM's own spawns already carry) plus a slop-ban list @@ -197,6 +223,21 @@ def _release_prompt( ) +def _revision_prompt( + rejected_body: str, reason: str, voice: str, product_name: str, context: str +) -> str: + extra = f"\n{context}\n" if context else "" + return ( + f"{voice}\n\n" + "The CEO rejected this draft tweet with feedback below. Revise it " + "into ONE new tweet (max 280 characters) that fully addresses the " + f"feedback for {product_name} — do not just repeat the rejected " + f"wording.\n{extra}\n" + f"Rejected draft:\n{rejected_body}\n\n" + f"CEO's feedback (address every point):\n{reason}\n" + ) + + def _reply_prompt(screened_mention_text: str, voice: str, product_name: str) -> str: return ( f"{voice}\n\n" @@ -1019,6 +1060,271 @@ class XEngine(BaseService): ) return task + # ---- reject -> redraft (CEO feedback loop) ----------------------------- + + async def redraft_from_rejection( + self, post_task: TaskTable, reason: str + ) -> TaskTable | None: + """Route a CEO's rejection reason into a fresh held draft of the SAME + source, mirroring ``VideoEngine.reauthor_from_rejection``. + + Reads the rejected draft's source-specific reference marker (release + version for x_post, mention ref for x_reply, feature ref for + x_feature) and asks the local model to revise the rejected body with + the CEO's reason folded in as guidance. Falls back gracefully: a + local-model failure or empty draft originates NOTHING — a degraded + copy is worse than none (see ``_draft_revision_body``). + + Deduped like ``VideoEngine.open_video_task``'s occasion scoping: a + redraft is skipped while one is already open for the same underlying + item (``_redraft_already_open``), so a repeated reject can't stack + drafts — but once that redraft is itself rejected (CANCELLED, so + excluded from ``list_open_x_posts``), a further redraft is allowed. + Also respects the shared open-post cap. The check+originate runs + under a per-identity Redis lock (``_acquire_redraft_lock`` — plain + SET NX, mirroring ``XPostService._acquire_lock``) so two deferred + closures racing for the SAME identity can't both pass the dedup + check and both originate; a held lock or an unreachable Redis skips + the redraft rather than risking a duplicate (the CEO can re-reject). + + Best-effort: never raises. Any failure (unresolvable project, + local-model error) logs a warning and returns None — the caller's + reject must succeed regardless of this seam. + """ + try: + return await self._redraft_from_rejection(post_task, reason) + except Exception as exc: + self.log.warning( + "x-engine: redraft from rejection failed", + task_id=str(post_task.id), + error=str(exc), + ) + return None + + async def _redraft_from_rejection( + self, post_task: TaskTable, reason: str + ) -> TaskTable | None: + if post_task.source not in X_SOURCES: + return None + identity = self._redraft_identity(post_task) + if identity is None: + # No stable discriminator to lock on (shouldn't happen for a real + # draft) — proceed unlocked, same as before this identity existed. + return await self._redraft_from_rejection_body(post_task, reason) + lock_key = f"{_REDRAFT_LOCK_PREFIX}{identity[0]}:{identity[1]}" + try: + token = await self._acquire_redraft_lock(lock_key) + except _RedraftLockUnavailable as exc: + self.log.warning( + "x-engine: redraft lock unavailable (redis down); skipping redraft", + task_id=str(post_task.id), + error=str(exc), + ) + return None + if token is None: + self.log.info( + "x-engine: a redraft is already in flight for this item; " + "skipping (the concurrent holder is already originating one)", + task_id=str(post_task.id), + ) + return None + try: + return await self._redraft_from_rejection_body(post_task, reason) + finally: + await self._release_redraft_lock(lock_key, token) + + async def _redraft_from_rejection_body( + self, post_task: TaskTable, reason: str + ) -> TaskTable | None: + """The check+originate critical section, run under the per-identity + redraft lock (or unlocked for an identity-less task — see caller).""" + open_posts = await get_task_service(self.session).list_open_x_posts() + if self._redraft_already_open(post_task, open_posts): + return None + if len(open_posts) >= settings.x_max_open_posts: + self.log.warning( + "x-engine: open-post cap reached; not redrafting rejected post", + task_id=str(post_task.id), + ) + return None + project = await self._resolve_redraft_project(post_task) + if project is None or project.id is None: + self.log.warning( + "x-engine: project not resolvable; skipping redraft", + task_id=str(post_task.id), + ) + return None + return await self._materialize_redraft(post_task, reason, project) + + async def _acquire_redraft_lock(self, lock_key: str) -> str | None: + """Non-blocking SET NX — mirrors XPostService._acquire_lock (same + style, a different key namespace).""" + token = uuid4().hex + try: + conn = redis.from_url(settings.redis_url) + try: + acquired = await conn.set( + lock_key, token, nx=True, ex=_REDRAFT_LOCK_TTL_SECONDS + ) + return token if acquired else None + finally: + await conn.aclose() + except Exception as exc: + raise _RedraftLockUnavailable(str(exc)) from exc + + async def _release_redraft_lock(self, lock_key: str, token: str) -> None: + try: + conn = redis.from_url(settings.redis_url) + try: + await conn.eval(_REDRAFT_RELEASE_SCRIPT, 1, lock_key, token) + finally: + await conn.aclose() + except Exception as exc: + self.log.warning( + "x-engine: redraft lock release failed (redis)", error=str(exc) + ) + + def _redraft_already_open( + self, post_task: TaskTable, open_posts: list[TaskTable] + ) -> bool: + """True when a redraft is already open for the SAME underlying item + (matched by ``_redraft_identity``) — the occasion-scoped dedup + ``open_video_task`` does for video, adapted to X's three sources. An + identity-less task (shouldn't happen for a real draft) never dedups.""" + identity = self._redraft_identity(post_task) + if identity is None: + return False + return any(self._redraft_identity(t) == identity for t in open_posts) + + def _redraft_identity(self, task: TaskTable) -> tuple[str, str] | None: + """(source, key) discriminating one draft's underlying item from + another of the same source: release version for x_post, mention id + for x_reply, feature slug for x_feature. None when the task carries + no such marker.""" + if task.source == X_POST_SOURCE: + version = markers.get_x_release_version(task) + return (X_POST_SOURCE, version) if version else None + if task.source == X_REPLY_SOURCE: + mention_id = (markers.get_x_mention_ref(task) or {}).get("id") + return (X_REPLY_SOURCE, str(mention_id)) if mention_id else None + if task.source == X_FEATURE_SOURCE: + slug = (markers.get_x_feature_ref(task) or {}).get("slug") + return (X_FEATURE_SOURCE, str(slug)) if slug else None + return None + + async def _resolve_redraft_project( + self, post_task: TaskTable + ) -> ProjectTable | None: + """The rejected draft's own project (every ``_originate_post`` call + sets one), or None when unresolvable.""" + if post_task.project_id is None: + return None + return await get_project_service(self.session).get( + cast("UUID", post_task.project_id) + ) + + async def _materialize_redraft( + self, post_task: TaskTable, reason: str, project: ProjectTable + ) -> TaskTable | None: + product_name = await get_company_goals_service( + self.session + ).resolve_product_name(project) + rejected_body = ( + markers.get_x_draft_body(post_task) or post_task.description or "" + ) + body = await self._draft_revision_body( + post_task=post_task, + rejected_body=rejected_body, + reason=reason, + product_name=product_name, + ) + if body is None: + self.log.info( + "x-engine: local model produced no revision; skipping redraft " + "(no degraded copy)", + task_id=str(post_task.id), + ) + return None + task = await self._originate_post( + title=post_task.title or "X post revision", + body=body, + source=post_task.source, + project_id=cast("UUID", project.id), + ) + self._carry_redraft_markers(task, post_task) + await self.session.flush() + self.log.info( + "x-engine: redraft opened after CEO rejection (held for CEO)", + rejected_task_id=str(post_task.id), + redraft_id=str(task.id), + ) + return task + + async def _draft_revision_body( + self, + *, + post_task: TaskTable, + rejected_body: str, + reason: str, + product_name: str, + ) -> str | None: + """The clamped revision body, or None when the local model produced + nothing usable — mirrors ``_draft_reply_body``'s no-degraded-copy + posture (see ``redraft_from_rejection``).""" + voice = await self._voice_guide(product_name) + context = self._redraft_context(post_task) + try: + draft = await _chat( + _revision_prompt(rejected_body, reason, voice, product_name, context) + ) + except Exception as exc: + self.log.warning( + "x-engine: local-model redraft failed (no degraded copy)", + task_id=str(post_task.id), + error=str(exc), + ) + return None + stripped = (draft or "").strip() + return _clamp_tweet(stripped) if stripped else None + + def _redraft_context(self, post_task: TaskTable) -> str: + """One line of source-specific factual grounding folded into the + revision prompt, so the redraft doesn't drift from what the post is + actually about.""" + if post_task.source == X_POST_SOURCE: + version = markers.get_x_release_version(post_task) + return ( + f"This is a release announcement for version {version}." + if version + else "" + ) + if post_task.source == X_REPLY_SOURCE: + text = (markers.get_x_mention_ref(post_task) or {}).get("text") + return f"This is a reply to this X mention:\n{text}" if text else "" + if post_task.source == X_FEATURE_SOURCE: + title = (markers.get_x_feature_ref(post_task) or {}).get("title") + return f"This is a feature-spotlight post about: {title}." if title else "" + return "" + + def _carry_redraft_markers(self, new_task: TaskTable, post_task: TaskTable) -> None: + """Copy the rejected draft's source-specific reference marker onto the + redraft so it renders identically in the panel queue/history and is + itself reject/approve-able through the normal flow. The feature + source's seen-slug bookkeeping is NOT repeated here — the original + draft already marked it seen at authoring time.""" + if post_task.source == X_POST_SOURCE: + version = markers.get_x_release_version(post_task) + if version: + markers.set_x_release_version(new_task, version) + elif post_task.source == X_REPLY_SOURCE: + ref = markers.get_x_mention_ref(post_task) + if ref: + markers.set_x_mention_ref(new_task, ref) + elif post_task.source == X_FEATURE_SOURCE: + ref = markers.get_x_feature_ref(post_task) + if ref: + markers.set_x_feature_ref(new_task, ref) + def get_x_engine(session: AsyncSession, client: XClient | None = None) -> XEngine: """Build an XEngine for ``session`` (optional injected client for tests).""" diff --git a/roboco/services/x_post_service.py b/roboco/services/x_post_service.py index 20ab0c5d..f15f3c0b 100644 --- a/roboco/services/x_post_service.py +++ b/roboco/services/x_post_service.py @@ -30,6 +30,7 @@ from roboco.config import settings from roboco.foundation.policy.content import markers from roboco.models.base import TaskStatus from roboco.services.base import BaseService +from roboco.services.notification_delivery import defer_after_commit from roboco.services.task import X_FEATURE_SOURCE, X_SOURCES, get_task_service from roboco.services.x_client import MAX_TWEET_CHARS, build_x_client from roboco.services.x_credentials import get_x_credentials_service @@ -253,7 +254,9 @@ class XPostService(BaseService): logger.warning("spotlight video draft failed (best-effort): %s", exc) async def reject(self, task_id: UUID, reason: str) -> TaskTable | None: - """Record the CEO's reason and cancel the draft (never posted). + """Record the CEO's reason, cancel the draft (never posted), and — for + a non-blank reason — schedule a redraft of the same source once this + transaction commits. Acquires the same post-mutex ``approve()`` holds (same key, same non-blocking acquire style) so a reject can't interleave with a @@ -263,40 +266,125 @@ class XPostService(BaseService): approve's own COMPLETED write right after it landed. Fails CLOSED like approve, both when the lock is held (approve mid-post) and when Redis is unreachable — the CEO retries the reject once it clears. + + Idempotent on an already-CANCELLED target — a stale/replayed reject + (e.g. a double-tapped Telegram button) returns the task UNCHANGED + rather than re-flushing the reason and, worse, scheduling ANOTHER + redraft; mirrors ``_approve_locked``'s ``already_rejected`` + short-circuit on the approve side. + + The redraft itself is scheduled via ``_schedule_redraft`` (see its + docstring for why it's deferred rather than inline) — a redraft + failure never fails or delays this call, and a blank/whitespace + reason schedules nothing (plain cancel semantics unchanged). """ task = await get_task_service(self.session).get(task_id) if task is None or task.source not in X_SOURCES: return None - if task.status == TaskStatus.COMPLETED: - raise TaskAlreadyCompletedError( - f"X draft {task_id} already posted (COMPLETED); cannot be rejected" - ) + terminal = self._reject_terminal_check(task, task_id) + if terminal is not None: + return terminal lock_key = f"{_LOCK_PREFIX}{task_id}" - try: - token = await self._acquire_lock(lock_key) - except _LockUnavailable as exc: - logger.error("x-post reject lock unavailable (redis down): %s", exc) - return None + token = await self._try_acquire_reject_lock(lock_key) if token is None: - return None # a concurrent approve is mid-post; refuse the reject + return None # redis down, or a concurrent approve is mid-post try: # Re-read under the lock: a concurrent approve may have posted + - # committed COMPLETED between the pre-lock check and here. + # committed COMPLETED between the pre-lock check and here, or a + # concurrent reject may have already landed CANCELLED. self.session.expire(task) locked = await get_task_service(self.session).get(task_id) if locked is None: return None - if locked.status == TaskStatus.COMPLETED: - raise TaskAlreadyCompletedError( - f"X draft {task_id} already posted (COMPLETED); cannot be rejected" - ) + terminal = self._reject_terminal_check(locked, task_id) + if terminal is not None: + return terminal markers.set_x_reject_reason(locked, reason) locked.status = TaskStatus.CANCELLED await self.session.flush() - return locked finally: await self._release_lock(lock_key, token) + # Outside the lock, after the cancel is flushed: a non-blank reason + # schedules the redraft. Never inline here (see _schedule_redraft). + if reason.strip(): + self._schedule_redraft(task_id, reason) + return locked + + def _reject_terminal_check( + self, task: TaskTable, task_id: UUID + ) -> TaskTable | None: + """Shared by the pre-lock and locked-under-mutex checks in + ``reject``: raises on COMPLETED (already posted, can't be + rejected), returns ``task`` unchanged on CANCELLED (an idempotent + replay — e.g. a double-tapped Telegram button — is a no-op, never + re-flushing the reason or scheduling ANOTHER redraft), else None to + mean "not terminal, proceed with the cancel".""" + if task.status == TaskStatus.COMPLETED: + raise TaskAlreadyCompletedError( + f"X draft {task_id} already posted (COMPLETED); cannot be rejected" + ) + if task.status == TaskStatus.CANCELLED: + return task + return None + + async def _try_acquire_reject_lock(self, lock_key: str) -> str | None: + """``_acquire_lock`` wrapped so a Redis-down failure and an + already-held lock look identical to the caller (both mean "can't + safely reject right now") — collapses two ``reject()`` return + sites (and the try/except) into one, under the complexity gate.""" + try: + return await self._acquire_lock(lock_key) + except _LockUnavailable as exc: + logger.error("x-post reject lock unavailable (redis down): %s", exc) + return None + + def _schedule_redraft(self, task_id: UUID, reason: str) -> None: + """Enqueue XEngine.redraft_from_rejection to run only after THIS + session's transaction actually commits (``defer_after_commit`` — + registers on ``self.session``'s own ``after_commit``/``after_rollback`` + events, so it fires regardless of whether the caller commits inside + this service call or, as the real route does, right after it + returns; a rollback drops it, never redrafting a reject that never + became durable). + + Deferred rather than inline for two reasons: ``reject`` is an HTTP + route handler and the local-model call inside the redraft can take + seconds — the response must not block on it; and by the time the + deferred callable actually runs, the request's own ``AsyncSession`` + may already be closing (FastAPI tears it down once the route + returns). So the callable opens a FRESH session from the process + session factory instead of touching ``self.session`` — the same + fresh-session-in-a-background-task shape ``TaskService. + _inject_proactive_context`` uses for its own post-commit background + write. Best-effort throughout: any failure (session error, project + unresolvable, local-model down) only logs. + """ + + async def _redraft() -> None: + from roboco.db.base import get_session_factory + from roboco.db.tables import TaskTable + from roboco.services.x_engine import get_x_engine + + try: + session_factory = get_session_factory() + async with session_factory() as session: + fresh = await session.get(TaskTable, task_id) + if fresh is None: + return + redrafted = await get_x_engine(session).redraft_from_rejection( + fresh, reason + ) + if redrafted is not None: + await session.commit() + except Exception as exc: # best-effort: never break the drain + logger.warning( + "x-post redraft-after-commit failed for task %s: %s", + task_id, + exc, + ) + + defer_after_commit(self.session, _redraft) # ---- Redis single-flight lock (plain SET NX — no heartbeat needed) ----- diff --git a/tests/unit/services/test_x_engine.py b/tests/unit/services/test_x_engine.py index b3fe1d07..835c12c6 100644 --- a/tests/unit/services/test_x_engine.py +++ b/tests/unit/services/test_x_engine.py @@ -8,9 +8,10 @@ posts — asserted against a real Postgres DB. from __future__ import annotations +from contextlib import contextmanager from datetime import UTC, datetime, timedelta from typing import TYPE_CHECKING, cast -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, patch from uuid import UUID, uuid4 import pytest @@ -47,6 +48,8 @@ from roboco.services.x_client import MAX_TWEET_CHARS, XClient, XMention, XPostRe from sqlalchemy import select if TYPE_CHECKING: + from collections.abc import Iterator + from sqlalchemy.ext.asyncio import AsyncSession SYSTEM_UUID = _foundation.AGENTS["system"].uuid @@ -1000,6 +1003,266 @@ async def test_materialize_feature_spotlight_enforces_280_chars( assert len(body) <= MAX_TWEET_CHARS +# --------------------------------------------------------------------------- # +# redraft_from_rejection — CEO reject feedback loop (mirrors VideoEngine. +# reauthor_from_rejection). ``post_task.status = TS.CANCELLED`` below mirrors +# XPostService.reject: by the time redraft_from_rejection runs, the rejected +# draft is already cancelled and excluded from list_open_x_posts. +# --------------------------------------------------------------------------- # + + +@contextmanager +def _redraft_lock_free() -> Iterator[None]: + """Patch XEngine's redraft-dedup lock helpers so a test exercises the + real check+originate path without touching the (test-blocked, see + conftest's `_no_live_redis`) Redis — mirrors test_x_post_service.py's + `_lock_free()` for the post-mutex. Every redraft test below that expects + the redraft to actually run needs this; the dedicated lock-held test + patches the SAME methods on the instance instead, to assert the opposite.""" + with ( + patch.object( + x_engine_module.XEngine, + "_acquire_redraft_lock", + AsyncMock(return_value="tok"), + ), + patch.object( + x_engine_module.XEngine, + "_release_redraft_lock", + AsyncMock(return_value=None), + ), + ): + yield + + +@pytest.mark.asyncio +async def test_redraft_from_rejection_release_post_carries_version_and_new_body( + db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch +) -> None: + await _seed(db_session) + _enable(monkeypatch) + _mock_local_model(monkeypatch, "shipped!") + engine = x_engine_module.XEngine(db_session, client=_FakeClient()) + post_task = await engine.draft_release_post( + version=_VERSION, highlights=["feat: x"] + ) + assert post_task is not None + original_body = markers.get_x_draft_body(post_task) + post_task.status = TS.CANCELLED + await db_session.flush() + + _mock_local_model(monkeypatch, "A sharper revised release announcement.") + with _redraft_lock_free(): + redraft = await engine.redraft_from_rejection( + post_task, "Too vague, name the feature" + ) + assert redraft is not None + assert redraft.id != post_task.id + assert redraft.source == X_POST_SOURCE + assert redraft.status == TS.PENDING + assert redraft.confirmed_by_human is False + assert markers.get_x_release_version(redraft) == _VERSION + body = markers.get_x_draft_body(redraft) + assert body == "A sharper revised release announcement." + assert body != original_body + + +@pytest.mark.asyncio +async def test_redraft_from_rejection_reply_carries_mention_ref( + db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch +) -> None: + await _seed(db_session) + _enable(monkeypatch) + _mock_local_model(monkeypatch, "reply") + engine = x_engine_module.XEngine( + db_session, client=_FakeClient(mentions=[_mention("m1", text="great work")]) + ) + result = await engine.run_cycle() + assert len(result) == ONE + post_task = result[0] + post_task.status = TS.CANCELLED + await db_session.flush() + + _mock_local_model(monkeypatch, "A better reply.") + with _redraft_lock_free(): + redraft = await engine.redraft_from_rejection(post_task, "Too generic") + assert redraft is not None + assert redraft.source == X_REPLY_SOURCE + ref = markers.get_x_mention_ref(redraft) + assert ref is not None + assert ref["id"] == "m1" + assert markers.get_x_draft_body(redraft) == "A better reply." + + +@pytest.mark.asyncio +async def test_redraft_from_rejection_feature_carries_feature_ref( + db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch +) -> None: + await _seed(db_session) + _enable(monkeypatch, x_feature_spotlight_enabled=True) + engine = x_engine_module.XEngine(db_session, client=_FakeClient()) + exploration = await engine.open_feature_spotlight_exploration() + assert exploration is not None + post_task = await engine.materialize_feature_spotlight( + exploration_task=exploration, + feature_slug="org-memory", + feature_title="Organizational Memory Loop", + body="Did you know RoboCo agents learn from every completed task?", + ) + post_task.status = TS.CANCELLED + await db_session.flush() + + _mock_local_model(monkeypatch, "A sharper spotlight tweet.") + with _redraft_lock_free(): + redraft = await engine.redraft_from_rejection(post_task, "Too dry") + assert redraft is not None + assert redraft.source == X_FEATURE_SOURCE + ref = markers.get_x_feature_ref(redraft) + assert ref is not None + assert ref["slug"] == "org-memory" + # No duplicate seen-slug insert — the original materialize already marked it. + assert await engine.is_feature_seen("org-memory") is True + + +@pytest.mark.asyncio +async def test_redraft_from_rejection_local_model_failure_originates_nothing( + db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch +) -> None: + """A local-model failure must never ship a degraded copy — no redraft.""" + await _seed(db_session) + _enable(monkeypatch) + _mock_local_model(monkeypatch, "shipped!") + engine = x_engine_module.XEngine(db_session, client=_FakeClient()) + post_task = await engine.draft_release_post(version=_VERSION, highlights=[]) + assert post_task is not None + post_task.status = TS.CANCELLED + await db_session.flush() + + monkeypatch.setattr( + x_engine_module, "_chat", AsyncMock(side_effect=RuntimeError("ollama down")) + ) + with _redraft_lock_free(): + redraft = await engine.redraft_from_rejection(post_task, "Needs work") + assert redraft is None + open_posts = await get_task_service(db_session).list_open_x_posts() + assert open_posts == [] + + +@pytest.mark.asyncio +async def test_redraft_from_rejection_empty_local_model_reply_originates_nothing( + db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch +) -> None: + await _seed(db_session) + _enable(monkeypatch) + _mock_local_model(monkeypatch, "shipped!") + engine = x_engine_module.XEngine(db_session, client=_FakeClient()) + post_task = await engine.draft_release_post(version=_VERSION, highlights=[]) + assert post_task is not None + post_task.status = TS.CANCELLED + await db_session.flush() + + _mock_local_model(monkeypatch, None) + with _redraft_lock_free(): + redraft = await engine.redraft_from_rejection(post_task, "Needs work") + assert redraft is None + + +@pytest.mark.asyncio +async def test_redraft_from_rejection_respects_open_cap( + db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch +) -> None: + await _seed(db_session) + _enable(monkeypatch, x_max_open_posts=1) + _mock_local_model(monkeypatch, "shipped!") + engine = x_engine_module.XEngine(db_session, client=_FakeClient()) + post_task = await engine.draft_release_post(version="1.0.0", highlights=[]) + assert post_task is not None + post_task.status = TS.CANCELLED # rejected; no longer counts toward the cap + await db_session.flush() + + # Fill the cap with an unrelated open draft (a mention reply). + filler_client = _FakeClient(mentions=[_mention("cap-fill")]) + filler_engine = x_engine_module.XEngine(db_session, client=filler_client) + filler = await filler_engine.run_cycle() + assert len(filler) == ONE + + with _redraft_lock_free(): + redraft = await engine.redraft_from_rejection(post_task, "Needs work") + assert redraft is None + + +@pytest.mark.asyncio +async def test_redraft_from_rejection_dedupes_while_open_then_allows_after_rejection( + db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch +) -> None: + """A redraft is skipped while one is already open for the same item — + mirroring VideoEngine's occasion-scoped dedup — so a repeated reject + can't stack unbounded drafts. But once THAT redraft is itself rejected + (CANCELLED, so excluded from list_open_x_posts), a further redraft + proceeds: a genuine second revision round still works.""" + await _seed(db_session) + _enable(monkeypatch) + _mock_local_model(monkeypatch, "shipped!") + engine = x_engine_module.XEngine(db_session, client=_FakeClient()) + post_task = await engine.draft_release_post(version=_VERSION, highlights=[]) + assert post_task is not None + post_task.status = TS.CANCELLED + await db_session.flush() + + with _redraft_lock_free(): + _mock_local_model(monkeypatch, "First revision.") + first_redraft = await engine.redraft_from_rejection( + post_task, "Round 1 feedback" + ) + assert first_redraft is not None + + # A second reject of the already-cancelled original must not stack + # another draft while the first redraft is still open. + again = await engine.redraft_from_rejection(post_task, "Round 1 feedback again") + assert again is None + + # Once the first redraft is itself rejected, a further redraft proceeds. + first_redraft.status = TS.CANCELLED + await db_session.flush() + _mock_local_model(monkeypatch, "Second revision.") + second_redraft = await engine.redraft_from_rejection( + first_redraft, "Round 2 feedback" + ) + assert second_redraft is not None + assert second_redraft.id != first_redraft.id + assert markers.get_x_draft_body(second_redraft) == "Second revision." + + +@pytest.mark.asyncio +async def test_redraft_from_rejection_skipped_when_lock_held( + db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch +) -> None: + """A concurrent redraft already holding the per-identity lock makes this + call skip rather than race the dedup check — closes the TOCTOU where two + deferred closures for the SAME identity could otherwise both pass + ``_redraft_already_open`` and both originate. A skipped redraft is safe: + the concurrent holder is already originating one, and the CEO can always + re-reject.""" + await _seed(db_session) + _enable(monkeypatch) + _mock_local_model(monkeypatch, "shipped!") + engine = x_engine_module.XEngine(db_session, client=_FakeClient()) + post_task = await engine.draft_release_post(version=_VERSION, highlights=[]) + assert post_task is not None + post_task.status = TS.CANCELLED + await db_session.flush() + + # Simulate a concurrent redraft already holding the lock for this identity. + release = AsyncMock() + monkeypatch.setattr(engine, "_acquire_redraft_lock", AsyncMock(return_value=None)) + monkeypatch.setattr(engine, "_release_redraft_lock", release) + + redraft = await engine.redraft_from_rejection(post_task, "Needs work") + assert redraft is None + release.assert_not_awaited() # never acquired, so never released + open_posts = await get_task_service(db_session).list_open_x_posts() + assert open_posts == [] + + # --------------------------------------------------------------------------- # # CHANGELOG.md parsing — pure functions, no DB/network needed # --------------------------------------------------------------------------- # diff --git a/tests/unit/services/test_x_post_service.py b/tests/unit/services/test_x_post_service.py index e7092692..5963f69a 100644 --- a/tests/unit/services/test_x_post_service.py +++ b/tests/unit/services/test_x_post_service.py @@ -7,11 +7,12 @@ fixture) so approve exercises the real post + status-transition path. from __future__ import annotations +import asyncio import contextlib from contextlib import contextmanager from typing import TYPE_CHECKING, cast from unittest.mock import AsyncMock, patch -from uuid import uuid4 +from uuid import UUID, uuid4 import pytest from roboco.config import settings as cfg @@ -27,6 +28,8 @@ from roboco.models.base import ( from roboco.models.base import TaskNature as TN from roboco.models.base import TaskStatus as TS from roboco.models.base import TaskType as TT +from roboco.services import x_engine as x_engine_module +from roboco.services.company_goals import get_company_goals_service from roboco.services.task import ( X_FEATURE_SOURCE, X_POST_SOURCE, @@ -41,6 +44,7 @@ from roboco.services.x_post_service import ( XPostService, get_x_post_service, ) +from sqlalchemy import delete from sqlalchemy.ext.asyncio import ( AsyncEngine, AsyncSession, @@ -50,7 +54,6 @@ from sqlalchemy.ext.asyncio import ( if TYPE_CHECKING: from collections.abc import Iterator - from uuid import UUID SYSTEM_UUID = _foundation.AGENTS["system"].uuid SECRETARY_UUID = _foundation.AGENTS["secretary-1"].uuid @@ -961,3 +964,295 @@ async def test_approve_feature_spotlight_video_failure_does_not_break_post( assert result.status == "posted" await db_session.refresh(task) assert task.status == TS.COMPLETED + + +# --------------------------------------------------------------------------- # +# Reject -> redraft (CEO feedback loop): a non-blank reject reason schedules +# XEngine.redraft_from_rejection to run only after this session's transaction +# actually commits (`defer_after_commit`), via a FRESH session opened from +# `get_session_factory()` — patched here to the same test database +# `db_session` uses, since the production singleton points elsewhere in +# tests. Mirrors `test_notification_delivery_phantom.py`'s drain helpers +# (duplicated locally rather than imported — this project's convention for +# a small pure test helper, not a service internal). +# --------------------------------------------------------------------------- # + + +def _drain_tasks(session: AsyncSession) -> list[asyncio.Task[object]]: + return list(session.info.get("_roboco_drain_tasks", [])) + + +async def _await_drain(session: AsyncSession) -> None: + tasks = _drain_tasks(session) + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + + +async def _redraft_engine_factory( + url: str, +) -> tuple[async_sessionmaker[AsyncSession], AsyncEngine]: + """A session factory on a brand-new engine bound to the SAME test + database `db_session` uses — what `_schedule_redraft`'s deferred closure + opens via `get_session_factory()` at drain time, patched here instead of + the (unreachable-in-tests) production singleton.""" + engine = create_async_engine(url, future=True) + factory = async_sessionmaker( + bind=engine, class_=AsyncSession, expire_on_commit=False + ) + return factory, engine + + +async def _delete_tasks(session: AsyncSession, *task_ids: UUID) -> None: + """Delete these task rows and commit. + + The tests below exercise a REAL `session.commit()` (required to fire the + after-commit redraft), and `_test_database_url` is a SESSION-scoped + database shared by every test in the whole run — an uncommitted row is + cleaned up by `db_session`'s own rollback-at-teardown, but a committed + one is durable and would leak an open draft into every later test (in + this file and any other) that counts/lists open X drafts. Explicit + cleanup restores the shared DB to a clean slate. + """ + await session.execute(delete(TaskTable).where(TaskTable.id.in_(task_ids))) + await session.commit() + + +@contextmanager +def _redraft_lock_free() -> Iterator[None]: + """Patch XEngine's redraft-dedup lock helpers (class-level, since the + deferred `_redraft()` closure constructs a fresh `XEngine` each time) so + the redraft's check+originate exercises its real path without touching + the (test-blocked) Redis — mirrors `_lock_free()` above for the post + mutex.""" + with ( + patch.object( + x_engine_module.XEngine, + "_acquire_redraft_lock", + AsyncMock(return_value="tok"), + ), + patch.object( + x_engine_module.XEngine, + "_release_redraft_lock", + AsyncMock(return_value=None), + ), + ): + yield + + +@pytest.mark.asyncio +async def test_reject_with_reason_schedules_deferred_redraft( + db_session: AsyncSession, +) -> None: + """A non-blank reason enqueues the redraft on the after-commit outbox — + nothing runs before the transaction actually commits.""" + task = await _seed_draft(db_session) + with _lock_free(): + await _svc(db_session).reject(_id(task), "Too vague") + assert db_session.info.get("_roboco_pending_bus_publishes") + + +@pytest.mark.asyncio +async def test_reject_blank_reason_schedules_no_redraft( + db_session: AsyncSession, +) -> None: + """Preserves current semantics: a blank/whitespace reason is a plain + cancel, nothing scheduled.""" + task = await _seed_draft(db_session) + with _lock_free(): + await _svc(db_session).reject(_id(task), " ") + assert not db_session.info.get("_roboco_pending_bus_publishes") + + +@pytest.mark.asyncio +async def test_reject_redraft_materializes_held_draft_of_same_source_with_new_body( + db_session: AsyncSession, + monkeypatch: pytest.MonkeyPatch, + _test_database_url: str, +) -> None: + """End to end: reject with a reason -> commit -> drain -> a fresh HELD + draft of the SAME source, carrying the local model's revised body. + + Pins the fresh-session contract two ways: `db_session` is CLOSED before + the drain runs (SQLAlchemy silently reopens a connection on reuse, so + this alone can't force a raise — it's included anyway per spec, and + still proves the drain doesn't NEED the request session kept open); the + real teeth is `get_x_engine` wrapped to capture the actual session + XEngine is constructed with and asserting it is NOT `db_session` — that + catches a "captured self.session instead of opening a fresh one" + regression regardless of `.close()`'s (non-)effect. + """ + # A non-empty brand_voice skips XEngine's one-time nudge notification — + # this test's own commit would otherwise durably flip that GLOBAL + # "already nudged" system_settings flag for the rest of the suite. + await get_company_goals_service(db_session).upsert( + {"brand_voice": "Confident, concise, no fluff."} + ) + monkeypatch.setattr( + x_engine_module, "_chat", AsyncMock(return_value="Revised body.") + ) + task = await _seed_draft(db_session, source=X_POST_SOURCE, body="Original body") + task_id = _id(task) + + factory, engine = await _redraft_engine_factory(_test_database_url) + monkeypatch.setattr("roboco.db.base.get_session_factory", lambda: factory) + + captured_sessions: list[AsyncSession] = [] + real_get_x_engine = x_engine_module.get_x_engine + + def _capturing_get_x_engine( + session: AsyncSession, client: XClient | None = None + ) -> x_engine_module.XEngine: + captured_sessions.append(session) + return real_get_x_engine(session, client=client) + + monkeypatch.setattr(x_engine_module, "get_x_engine", _capturing_get_x_engine) + + with _redraft_lock_free(): + try: + with _lock_free(): + await _svc(db_session).reject(task_id, "Needs a concrete detail") + await db_session.commit() + await db_session.close() + await _await_drain(db_session) + finally: + await engine.dispose() + + assert len(captured_sessions) == 1 + assert captured_sessions[0] is not db_session + + open_posts = await _svc(db_session).list_open_posts() + redrafts = [t for t in open_posts if t.id != task_id] + assert len(redrafts) == 1 + redraft = redrafts[0] + assert redraft.source == X_POST_SOURCE + assert redraft.status == TS.PENDING + assert redraft.confirmed_by_human is False + assert markers.get_x_draft_body(redraft) == "Revised body." + + await _delete_tasks(db_session, task_id, UUID(str(redraft.id))) + + +@pytest.mark.asyncio +async def test_reject_redraft_local_model_failure_originates_nothing( + db_session: AsyncSession, + monkeypatch: pytest.MonkeyPatch, + _test_database_url: str, +) -> None: + """A local-model failure at redraft time never ships a degraded copy — + the reject stays a plain cancel with no fresh draft.""" + monkeypatch.setattr( + x_engine_module, "_chat", AsyncMock(side_effect=RuntimeError("ollama down")) + ) + task = await _seed_draft(db_session, source=X_POST_SOURCE) + task_id = _id(task) + + factory, engine = await _redraft_engine_factory(_test_database_url) + monkeypatch.setattr("roboco.db.base.get_session_factory", lambda: factory) + with _redraft_lock_free(): + try: + with _lock_free(): + updated = await _svc(db_session).reject(task_id, "Needs work") + assert updated is not None + assert updated.status == TS.CANCELLED + await db_session.commit() + await _await_drain(db_session) + finally: + await engine.dispose() + + open_posts = await _svc(db_session).list_open_posts() + assert open_posts == [] + + await _delete_tasks(db_session, task_id) + + +@pytest.mark.asyncio +async def test_reject_redraft_respects_open_post_cap( + db_session: AsyncSession, + monkeypatch: pytest.MonkeyPatch, + _test_database_url: str, +) -> None: + """The shared open-post cap blocks the redraft exactly like it blocks any + other origination — an unrelated open draft filling the cap means the + rejected item gets no revision this cycle.""" + monkeypatch.setattr(cfg, "x_max_open_posts", 1) + monkeypatch.setattr( + x_engine_module, "_chat", AsyncMock(return_value="Revised body.") + ) + task = await _seed_draft(db_session, source=X_POST_SOURCE) + task_id = _id(task) + filler = await _seed_draft(db_session, source=X_REPLY_SOURCE, body="Filler reply") + + factory, engine = await _redraft_engine_factory(_test_database_url) + monkeypatch.setattr("roboco.db.base.get_session_factory", lambda: factory) + with _redraft_lock_free(): + try: + with _lock_free(): + await _svc(db_session).reject(task_id, "Needs a concrete detail") + await db_session.commit() + await _await_drain(db_session) + finally: + await engine.dispose() + + open_posts = await _svc(db_session).list_open_posts() + ids = {t.id for t in open_posts} + assert ids == {filler.id} # no redraft — the cap was already at 1 + + await _delete_tasks(db_session, task_id, UUID(str(filler.id))) + + +@pytest.mark.asyncio +async def test_reject_survives_deferred_session_factory_failure( + db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch +) -> None: + """A redraft failure at drain time (here: the fresh-session open itself + blowing up) must never break the reject — it already committed CANCELLED + before this best-effort seam ever runs.""" + task = await _seed_draft(db_session, source=X_POST_SOURCE) + task_id = _id(task) + + def _boom() -> async_sessionmaker[AsyncSession]: + raise RuntimeError("db down") + + monkeypatch.setattr("roboco.db.base.get_session_factory", _boom) + with _lock_free(): + updated = await _svc(db_session).reject(task_id, "Needs work") + assert updated is not None + assert updated.status == TS.CANCELLED + + await db_session.commit() + await _await_drain(db_session) # must not raise + + await db_session.refresh(updated) + assert updated.status == TS.CANCELLED + assert markers.get_x_reject_reason(updated) == "Needs work" + + await _delete_tasks(db_session, task_id) + + +@pytest.mark.asyncio +async def test_reject_replayed_on_already_cancelled_is_noop( + db_session: AsyncSession, +) -> None: + """A second reject() on an already-CANCELLED task (a stale/replayed + request — e.g. a double-tapped Telegram button) is idempotent: it + returns the task UNCHANGED, never re-flushes the reason, and schedules + NO second redraft — mirroring approve()'s already_rejected + short-circuit. Pre-fix this re-flushed CANCELLED and scheduled another + redraft on every replay.""" + task = await _seed_draft(db_session, source=X_POST_SOURCE) + task_id = _id(task) + with _lock_free(): + first = await _svc(db_session).reject(task_id, "Needs work") + assert first is not None + assert first.status == TS.CANCELLED + # The first reject scheduled its own redraft — clear the pending queue + # so the assertion below is unambiguous about what the SECOND call does. + db_session.info.pop("_roboco_pending_bus_publishes", None) + + with _lock_free(): + second = await _svc(db_session).reject(task_id, "A completely different reason") + assert second is not None + assert second.status == TS.CANCELLED + # Unchanged: the replay's reason must NOT overwrite the original. + assert markers.get_x_reject_reason(second) == "Needs work" + assert not db_session.info.get("_roboco_pending_bus_publishes")