feat(x): redraft loop on CEO reject — feedback re-enters the draft flow (#648)

* 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 <rennf93@users.noreply.github.com>

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)."""

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:
+        terminal = self._reject_terminal_check(task, task_id)
-            raise TaskAlreadyCompletedError(
+        if terminal is not None:
-                f"X draft {task_id} already posted (COMPLETED); cannot be rejected"
+            return terminal
-            )
 
         lock_key = f"{_LOCK_PREFIX}{task_id}"
-        try:
+        token = await self._try_acquire_reject_lock(lock_key)
-            token = await self._acquire_lock(lock_key)
-        except _LockUnavailable as exc:
-            logger.error("x-post reject lock unavailable (redis down): %s", exc)
-            return None
         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:
+            terminal = self._reject_terminal_check(locked, task_id)
-                raise TaskAlreadyCompletedError(
+            if terminal is not None:
-                    f"X draft {task_id} already posted (COMPLETED); cannot be rejected"
+                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) -----
 

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
 # --------------------------------------------------------------------------- #

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")