Continue agents after workflow transitions

AGENTS.md

@@ -264,8 +264,9 @@ on every PR (the same commands the Makefile wraps).
   commands): `advance` is the **happy path** — auto-derived as a state's single non-`DROPPED`
   declared transition (gated by responsibilities) — and `drop` (→ `DROPPED`) is the universal
   escape. Those are the core operations (a workflow may declare more, but each must target a
-  legal transition). `advance` starts a new agentic turn, so it's invoked by an **in-container
-  agent skill** (over REST/MCP); the dashboard drives only `drop` (`x`).
+  legal transition). `advance` is invoked by an **in-container agent skill** (over REST/MCP);
+  its tool result carries the entered phase's briefing so the task's agent continues immediately
+  when it still holds the turn. The dashboard drives only `drop` (`x`).
 - **Free move / set state** — moving a task to *any* state directly (`set_state` /
   `PUT …/state`), bypassing the declared graph **and** the responsibility gate. A workflow's
   `transitions` declare only the intended path (what `advance` follows); the user is never boxed

docs/tasks.md

@@ -104,8 +104,9 @@ Three ways a task changes state:
 
 - **`advance`** — the happy path. A state with a single non-`DROPPED` transition derives an
   `advance` operation automatically; taking it is **gated on the state's responsibilities**
-  all being resolved. This starts a new agentic turn, so an in-container agent skill invokes
-  it (over MCP), never the dashboard.
+  all being resolved. An in-container agent skill invokes it (over MCP), never the dashboard;
+  the result carries the entered phase's briefing so the task's agent can continue immediately
+  when it holds that phase's turn.
 - **`drop`** — the universal escape. Every non-terminal state can go straight to `DROPPED`
   (dashboard `x`), no gate. Nothing lands.
 - **free move (`set_state`)** — moving a task to *any* state directly, off the declared

src/panopticon/container/skills.py

@@ -46,14 +46,30 @@ def render_operation(name: str, target_state: str, task_id: str) -> str:
 
     CLI-agnostic — both adapters write this same text; only the destination dir differs.
     Operations are the workflow's **declared, gated** moves; the agent applies one by name via the
-    `apply_operation` tool (not by editing state directly), which starts a new agentic turn.
+    `apply_operation` tool (not by editing state directly), then follows the returned briefing.
     """
     return (
         f"---\ndescription: Apply the workflow's '{name}' operation.\n---\n"
+        f"{_operation_body(name, target_state, task_id)}"
+    )
+
+
+def _operation_body(name: str, target_state: str, task_id: str) -> str:
+    """Shared operation procedure for the Claude and Codex rendering surfaces."""
+    invocation = (
+        f'Invoke the `apply_operation` tool with `operation="{name}"`, `task_id="{task_id}"`, '
+        f'and `acting_task="{task_id}"`; don\'t edit the state directly. '
+    )
+    if name == "drop":
+        detail = "Dropping is always allowed and bypasses outstanding responsibilities."
+    else:
+        detail = (
+            "The operation is gated on the current state's responsibilities. Follow the entered "
+            "phase's briefing returned by the tool."
+        )
+    return (
         f"Apply this workflow's `{name}` operation — it moves the task to **{target_state}**. "
-        f'Invoke it with the `apply_operation` tool (`operation="{name}"`, `task_id="{task_id}"`); '
-        f"don't edit the state directly. It's gated on the current state's responsibilities and "
-        f"starts a new turn.\n"
+        f"{invocation}{detail}\n"
     )
 
 
@@ -113,10 +129,7 @@ def render_agent_operation(name: str, target_state: str, task_id: str) -> str:
     """The rendered ``SKILL.md`` body for a core operation on the codex agent-skills surface."""
     return (
         f"---\nname: {name}\ndescription: Apply the workflow's '{name}' operation.\n---\n"
-        f"Apply this workflow's `{name}` operation — it moves the task to **{target_state}**. "
-        f'Invoke it with the `apply_operation` tool (`operation="{name}"`, `task_id="{task_id}"`); '
-        f"don't edit the state directly. It's gated on the current state's responsibilities and "
-        f"starts a new turn.\n"
+        f"{_operation_body(name, target_state, task_id)}"
     )
 
 

src/panopticon/taskservice/mcp.py

@@ -17,7 +17,7 @@
 from mcp.server.transport_security import TransportSecuritySettings
 
 from panopticon.core.artifacts import decode_b64_artifact, decode_segment, mcp_uri
-from panopticon.core.models import Actor, Status
+from panopticon.core.models import Actor, Status, Task
 from panopticon.taskservice.api import TaskOut
 from panopticon.taskservice.service import TaskService
 
@@ -32,6 +32,32 @@ def _task(task: object) -> dict[str, Any]:
     return TaskOut.model_validate(task).model_dump(mode="json")
 
 
+async def _transition_result(
+    service: TaskService, task: Task, *, acting_task: str | None
+) -> dict[str, Any]:
+    """Serialize a transition and put the entered phase's briefing in the tool result.
+
+    A transition happens mid-model-turn, after the user-prompt hook emitted the old phase's
+    briefing. Returning the new one here closes that context gap for every MCP-capable agent CLI.
+    """
+    result = _task(task)
+    try:
+        briefing = await service.briefing_for(task)
+    except Exception:  # The persisted transition must not look like a failed tool call.
+        _log.exception("task %s transitioned but its briefing could not be rendered", task.id)
+        return result
+    if (
+        task.turn is Actor.AGENT
+        and not service.is_terminal(task)
+        and (acting_task is None or acting_task == task.id)
+    ):
+        briefing += (
+            "\n\nYou now hold the turn in this phase. Continue its work immediately in this "
+            "same turn; do not stop merely to wait for another user prompt."
+        )
+    return {**result, "briefing": briefing}
+
+
 def build_mcp_server(service: TaskService, *, name: str = "panopticon") -> FastMCP:
     """An MCP server exposing the task service's agent-facing operations + artifacts."""
     # Disable the SDK's DNS-rebinding (Host/Origin) guard: the agent reaches us across the
@@ -56,14 +82,18 @@ async def set_url(task_id: str, url: str) -> dict[str, Any]:
         return _task(await service.set_url(task_id, url))
 
     @mcp.tool(description="Apply a named core operation (e.g. 'advance', 'drop').")
-    async def apply_operation(task_id: str, operation: str) -> dict[str, Any]:
+    async def apply_operation(
+        task_id: str, operation: str, acting_task: str | None = None
+    ) -> dict[str, Any]:
         _log.debug("mcp apply_operation task=%s operation=%s", task_id, operation)
-        return _task(await service.apply_operation(task_id, operation))
+        task = await service.apply_operation(task_id, operation)
+        return await _transition_result(service, task, acting_task=acting_task)
 
     @mcp.tool(description="Move the task to any state directly (free move; bypasses the gate).")
-    async def set_state(task_id: str, state: str) -> dict[str, Any]:
+    async def set_state(task_id: str, state: str, acting_task: str | None = None) -> dict[str, Any]:
         _log.debug("mcp set_state task=%s state=%s", task_id, state)
-        return _task(await service.set_state(task_id, state))
+        task = await service.set_state(task_id, state)
+        return await _transition_result(service, task, acting_task=acting_task)
 
     @mcp.tool(
         description="Resolve one promised responsibility ('met', or 'failed' with a comment)."

src/panopticon/taskservice/service.py

@@ -493,8 +493,21 @@ async def briefing(self, task_id: str) -> str:
         """A short briefing on the task's current phase (state + responsibilities + how it advances),
         rendered from the workflow so the in-container agent knows *where it is* (the hook emits it)."""
         task = await self.get_task(task_id)
+        return await self.briefing_for(task)
+
+    async def briefing_for(self, task: Task) -> str:
+        """Render the briefing for an already-loaded task.
+
+        Transition callers use this with the exact post-transition object so their response cannot
+        race a second task lookup. The user-prompt hook's :meth:`briefing` path delegates here too,
+        keeping one source for phase, responsibility, and artifact-aware briefing text.
+        """
         return await self._workflow(task.workflow).briefing(task, artifacts=self._artifacts)
 
+    def is_terminal(self, task: Task) -> bool:
+        """Whether an already-loaded task is in a terminal state of its workflow."""
+        return self._workflow(task.workflow).is_terminal(task.state)
+
     async def workflow_overview(self, task_id: str) -> str:
         """A one-time map of the task's whole workflow (the agent gets this in its system prompt)."""
         task = await self.get_task(task_id)

src/panopticon/terminal/dashboard.py

@@ -26,9 +26,9 @@
 opens the on-disk file in place when the dashboard shares the artifact store, `y` **copies the
 task's slug** and `Y` its **id** to the clipboard (OSC 52 + the host's `pbcopy`/`xclip`/`wl-copy`,
 so it works on Linux and macOS). Drop is the only state
-*transition* the dashboard drives: every other transition starts a new agentic turn, so it's
-triggered by an in-container agent skill (`advance` over REST/MCP; going back to coding is a free
-`set_state` move), not the operator (ADR 0004).
+*transition* the dashboard drives: every other transition is triggered by an in-container agent
+skill (`advance` over REST/MCP; going back to coding is a free `set_state` move), and its MCP result
+briefs the agent on the entered phase (ADR 0004).
 
 `/` enters **search-as-you-type** (cloude-cade's `/`): a query box reveals at the bottom and the
 table filters live to tasks whose slug/state/workflow/memo contains the query
@@ -2289,8 +2289,8 @@ def create(result: tuple[str, bool, dict[str, str]] | None) -> None:
 
     def action_drop(self) -> None:
         """`x`: abandon the highlighted task. Drop is the **only** transition the dashboard
-        drives — every other transition starts a new agentic turn, so it's triggered by an
-        in-container agent skill, not the operator (ADR 0004)."""
+        drives — every other transition is triggered by an in-container agent skill and returns
+        the entered phase's briefing, not by the operator (ADR 0004)."""
         task_id = self._current
         if task_id is None:
             return

tests/container/test_skills.py

@@ -4,7 +4,12 @@
 
 from pathlib import Path
 
-from panopticon.container.skills import render_command, render_operation, write_commands
+from panopticon.container.skills import (
+    render_agent_operation,
+    render_command,
+    render_operation,
+    write_commands,
+)
 from panopticon.core.models import Skill
 
 
@@ -21,6 +26,28 @@ def test_render_operation_injects_the_task_id() -> None:
     body = render_operation("advance", "COMPLETE", "t-9")
     assert "apply_operation" in body and "COMPLETE" in body
     assert 'operation="advance"' in body and 'task_id="t-9"' in body
+    assert 'acting_task="t-9"' in body
+    assert "Follow the entered phase's briefing" in body
+    assert "starts a new turn" not in body
+
+
+def test_render_agent_operation_tells_codex_to_follow_the_returned_briefing() -> None:
+    body = render_agent_operation("advance", "ITERATING", "t-9")
+    assert "Follow the entered phase's briefing" in body
+    assert "starts a new turn" not in body
+
+
+def test_operation_renderers_share_one_procedure_body() -> None:
+    claude = render_operation("advance", "ITERATING", "t-9").split("---\n", 2)[2]
+    codex = render_agent_operation("advance", "ITERATING", "t-9").split("---\n", 2)[2]
+    assert claude == codex
+
+
+def test_drop_operation_is_ungated_and_has_no_continue_guidance() -> None:
+    body = render_operation("drop", "DROPPED", "t-9")
+    assert "always allowed and bypasses outstanding responsibilities" in body
+    assert "gated" not in body
+    assert "briefing" not in body
 
 
 def test_write_commands_writes_one_file_per_skill(tmp_path: Path) -> None:

tests/taskservice/test_mcp.py

@@ -5,10 +5,11 @@
 
 import base64
 from pathlib import Path
+from unittest.mock import AsyncMock
 
 from mcp.shared.memory import create_connected_server_and_client_session as connect
 
-from panopticon.core.models import Actor, Repo
+from panopticon.core.models import Actor, Repo, Status
 from panopticon.taskservice.artifacts_fs import FilesystemArtifactStore
 from panopticon.taskservice.mcp import build_mcp_server
 from panopticon.taskservice.service import TaskService
@@ -65,13 +66,91 @@ async def test_tools_are_exposed_and_drive_the_task(tmp_path: Path) -> None:
             "put_artifact",
             "list_artifacts",
         } <= names
-        result = await s.call_tool("apply_operation", {"task_id": task.id, "operation": "advance"})
+        result = await s.call_tool(
+            "apply_operation",
+            {"task_id": task.id, "operation": "advance", "acting_task": task.id},
+        )
         assert result.isError is False
         assert result.structuredContent is not None
         assert result.structuredContent["state"] == "COMPLETE"
+        briefing = result.structuredContent["briefing"]
+        assert "terminal state **COMPLETE**" in briefing
+        assert "Continue its work immediately" not in briefing
     assert (await svc.get_task(task.id)).state == "COMPLETE"  # the tool actually mutated the task
 
 
+async def test_advance_returns_the_entered_phase_briefing_and_continue_directive(
+    tmp_path: Path,
+) -> None:
+    svc = await _service(tmp_path)
+    task = await svc.create_task("r1", "github-self-reviewed")
+    await svc.put_artifact(task.id, "plan.md", b"# Build the widget")
+    await svc.resolve_responsibility(task.id, "plan-written", status=Status.MET)
+
+    async with connect(build_mcp_server(svc)) as s:
+        await s.initialize()
+        result = await s.call_tool("apply_operation", {"task_id": task.id, "operation": "advance"})
+
+    assert result.structuredContent is not None
+    assert result.structuredContent["state"] == "ITERATING"  # existing flat task shape remains
+    briefing = result.structuredContent["briefing"]
+    assert "**ITERATING**" in briefing
+    assert f"panopticon://tasks/{task.id}/artifacts/plan.md" in briefing
+    assert "Continue its work immediately in this same turn" in briefing
+
+
+async def test_set_state_returns_the_entered_phase_briefing(tmp_path: Path) -> None:
+    svc = await _service(tmp_path)
+    task = await svc.create_task("r1", "github-self-reviewed")
+
+    async with connect(build_mcp_server(svc)) as s:
+        await s.initialize()
+        result = await s.call_tool("set_state", {"task_id": task.id, "state": "ITERATING"})
+
+    assert result.structuredContent is not None
+    assert result.structuredContent["state"] == "ITERATING"
+    briefing = result.structuredContent["briefing"]
+    assert "**ITERATING**" in briefing
+    assert "Continue its work immediately in this same turn" in briefing
+
+
+async def test_cross_task_transition_does_not_direct_the_caller_to_do_the_targets_work(
+    tmp_path: Path,
+) -> None:
+    svc = await _service(tmp_path)
+    acting = await svc.create_task("r1", "orchestrator")
+    target = await svc.create_task("r1", "github-self-reviewed")
+
+    async with connect(build_mcp_server(svc)) as s:
+        await s.initialize()
+        result = await s.call_tool(
+            "set_state",
+            {"task_id": target.id, "state": "ITERATING", "acting_task": acting.id},
+        )
+
+    assert result.structuredContent is not None
+    assert "**ITERATING**" in result.structuredContent["briefing"]
+    assert "Continue its work immediately" not in result.structuredContent["briefing"]
+
+
+async def test_briefing_failure_does_not_make_a_persisted_transition_look_failed(
+    tmp_path: Path,
+) -> None:
+    svc = await _service(tmp_path)
+    task = await svc.create_task("r1", "spike")
+    svc.briefing_for = AsyncMock(side_effect=OSError("artifact store unavailable"))  # type: ignore[method-assign]
+
+    async with connect(build_mcp_server(svc)) as s:
+        await s.initialize()
+        result = await s.call_tool("apply_operation", {"task_id": task.id, "operation": "advance"})
+
+    assert result.isError is False
+    assert result.structuredContent is not None
+    assert result.structuredContent["state"] == "COMPLETE"
+    assert "briefing" not in result.structuredContent
+    assert (await svc.get_task(task.id)).state == "COMPLETE"
+
+
 async def test_artifacts_round_trip_via_tool_and_resource(tmp_path: Path) -> None:
     svc = await _service(tmp_path)
     task = await svc.create_task("r1", "spike")