feat(progress): plan-driven progress — % derived from the plan checklist (#173)

Progress was only the synthetic milestone entry (auto-emitted at
open_pr/i_am_done); agents never deliberately reported and the % was
an ungated free-form guess.

Now the plan's sub_tasks ARE the progress skeleton:
- progress() gains optional `plan_step` (a sub_task id or its 1-based
  order). With it, that step is marked completed and the percentage is
  DERIVED as completed/total (equal weight) via new
  TaskService.record_plan_progress — the agent cannot set/game it.
- A narrative entry WITHOUT plan_step is allowed for important
  mid-step documentation and carries the current derived % (the bar
  never regresses). No hard anti-spam gate (would loop minimax) —
  prompt guidance steers "meaningful moments, not every tool call".
- `percentage` is now an optional fallback, used only for tasks with
  no sub_task checklist (back-compat). v2 ProgressRequest, the do.py
  route, and the do_server MCP tool updated accordingly.
- An unmatched plan_step returns invalid_state listing the valid step
  refs (resolve by id / order / 1-based index).
- developer + documenter prompts updated to the plan_step workflow.
- Helpers extracted (_plan_subtasks/_derive_plan_pct/_valid_step_refs/
  _mark_subtask_complete) to keep record_plan_progress within the
  cyclomatic gate.

Commit 3 of 3 for the plan/progress quality work (#171/#172/#173).

agents/prompts/roles/developer.md

@@ -59,7 +59,7 @@ When you respawn, your task is in some lifecycle status. The next call follows f
 3. `note(scope='decision', text='<approach: files I'll touch, plan, risks, how I'll verify each criterion>')` -> records your reasoning before claiming.
 4. `i_will_work_on(task_id, plan="<scope, files, approach, risks>")` -> claims, creates branch, sets `in_progress`.
 5. Edit / Write your changes inside the workspace. Run tests via `Bash` after each meaningful change.
-6. `commit(message=...)` after each meaningful change. Then `progress(task_id, message="<one sentence about what just landed>", percentage=<0..100>)` to surface narrative progress to the Plan/Progress tab. Commits are git refs; progress is the human-readable update for QA / PM / CEO. Repeat 5-6 until the criteria are met.
+6. `commit(message=...)` after each meaningful change. **As you FINISH each plan step, call `progress(task_id, plan_step="<that step's id or 1-based order>", message="<one sentence about what landed>")`** — the step is marked complete and the % is computed from your checklist for you (do NOT pass `percentage`; you cannot set it). You MAY also post a `progress(task_id, message=...)` WITHOUT `plan_step` for an important mid-step milestone (documents the "why"; carries the current %) — meaningful moments only, not every tool call. Commits are git refs; progress maps to your plan for QA / PM / CEO. Repeat 5-6 until every step is done and the criteria are met.
 7. If you get stuck (test won't pass, design unclear, deps missing): `note(scope='struggle', text='<what's stuck + what you've tried>')` BEFORE moving to `i_am_blocked`. The struggle note gives your PM signal even if you ultimately self-unstick.
 8. When a struggle resolves: `note(scope='learning', text='<what worked + why>')` so the next agent benefits.
 9. `note(scope='reflect', text="<what you did + why + how each acceptance criterion was met>")` before submitting. **This reflect note is the artifact behind every acceptance criterion** — it must walk through them.

agents/prompts/roles/documenter.md

@@ -29,7 +29,7 @@ You do NOT re-implement the developer's work. You do NOT review or critique the
 | `evidence(task_id)` | Re-fetches PR diff and commits if needed. | None. |
 | `roboco_git_status(project_slug)` / `roboco_git_log(project_slug, limit?, branch?)` / `roboco_git_diff(project_slug, branch?, base?)` / `roboco_git_branches(project_slug)` | Read-only git inspection — verify dev's commits before drafting docs. | None. |
 | `i_am_idle()` | Done for now. Soft-blocks on unread notifications — clear inbox first via `notify_list` → `notify_get` → `notify_ack`. | No active doc claim. |
-| `progress(task_id, message, percentage)` | Append a narrative progress entry to the panel's Progress tab (0..100). Use in addition to `commit()`. **NOT `TodoWrite`** — TodoWrite is your private session scratchpad that does NOT surface to the panel. | Task assigned to you and active. |
+| `progress(task_id, message, plan_step?)` | Record progress to the panel's Progress tab. Pass `plan_step` (a sub_task id or its 1-based order) as you finish each plan step — it is marked complete and the % is **computed for you** (you do not set `percentage`). A narrative entry without `plan_step` is allowed for an important milestone. Use in addition to `commit()`. **NOT `TodoWrite`** — TodoWrite is your private session scratchpad that does NOT surface to the panel. | Task assigned to you and active. |
 | `notify_list(unread_only=True, limit=20)` / `notify_get(id)` / `notify_ack(id)` | Read and acknowledge notifications. | None. |
 
 ## State → Verb

roboco/api/routes/v2/do.py

@@ -151,6 +151,7 @@ async def do_progress(
         agent_id=x_agent_id,
         task_id=body.task_id,
         message=body.message,
+        plan_step=body.plan_step,
         percentage=body.percentage,
     )
     return envelope_to_response(env, request)

roboco/api/schemas/v2/do.py

@@ -102,15 +102,23 @@ class LinkSessionRequest(BaseModel):
 
 
 class ProgressRequest(BaseModel):
-    """Narrative progress update with 0..100 percentage.
+    """Progress update; % is DERIVED from the plan checklist (#173).
 
-    Pre-gateway parity for `roboco_task_progress`. Populates the panel's
+    Pass ``plan_step`` (a sub_task id or its 1-based order) as you finish
-    Progress tab.
+    each plan step — it is marked complete and the percentage is computed
+    from completed/total. A narrative entry without ``plan_step`` is
+    allowed for important mid-step documentation. ``percentage`` is an
+    optional fallback only for tasks with no sub_task checklist.
+    Populates the panel's Progress tab.
     """
 
     task_id: UUID
     message: str = Field(..., min_length=1)
-    percentage: int = Field(..., ge=0, le=100)
+    plan_step: str | None = Field(
+        default=None,
+        description="sub_task id or 1-based order to mark complete",
+    )
+    percentage: int | None = Field(default=None, ge=0, le=100)
 
 
 class NotifyListRequest(BaseModel):

roboco/mcp/do_server.py

@@ -303,22 +303,42 @@ def evidence(task_id: str) -> dict[str, Any]:
 # ---------- Wave 1 — pre-gateway parity ----------
 
 
-def progress(task_id: str, message: str, percentage: int) -> dict[str, Any]:
+def progress(
-    """Append a narrative progress update to YOUR active task.
+    task_id: str,
+    message: str,
+    plan_step: str | None = None,
+    percentage: int | None = None,
+) -> dict[str, Any]:
+    """Record progress on YOUR active task — the % is computed for you.
+
+    Your plan's steps (sub_tasks) ARE the progress checklist. As you
+    FINISH each step, call this with ``plan_step`` set to that step's id
+    or its 1-based order; it is marked complete and the percentage is
+    derived from completed/total — you do NOT set the percentage and
+    cannot game it.
+
+    You may ALSO post a narrative update WITHOUT ``plan_step`` for an
+    important mid-step milestone (it documents the "why" and carries the
+    current derived %). Keep these to meaningful moments — not every
+    tool call.
 
     Args:
         task_id: UUID of the task you're working on.
         message: One-paragraph summary of what just landed.
-        percentage: 0..100 inclusive. Rough completion estimate; bump it as
+        plan_step: The sub_task id (or its 1-based order) you just
-            you make progress so PM/QA can see velocity.
+            COMPLETED. Omit for a narrative-only milestone update.
+        percentage: Ignored when the task has a plan checklist (the norm).
+            Only used as a fallback for tasks with no sub_tasks.
 
-    Populates the panel's Progress tab. Use this in addition to ``commit``
+    Populates the panel's Progress tab. Use in addition to ``commit`` —
-    — commits are git refs; progress is narrative.
+    commits are git refs; progress maps to your plan.
     """
-    return _post(
+    body: dict[str, Any] = {"task_id": task_id, "message": message}
-        "/api/v2/do/progress",
+    if plan_step is not None:
-        {"task_id": task_id, "message": message, "percentage": percentage},
+        body["plan_step"] = plan_step
-    )
+    if percentage is not None:
+        body["percentage"] = percentage
+    return _post("/api/v2/do/progress", body)
 
 
 def open_session(

roboco/services/gateway/content_actions.py

@@ -713,13 +713,20 @@ class ContentActions:
         agent_id: UUID,
         task_id: UUID,
         message: str,
-        percentage: int,
+        plan_step: str | None = None,
+        percentage: int | None = None,
     ) -> Envelope:
-        """Append a narrative progress update (pre-gateway parity).
+        """Append a progress update; % is derived from the plan checklist.
+
+        #173: pass ``plan_step`` (a sub_task id or 1-based order) as you
+        finish each plan step — it is marked complete and the % is
+        computed from completed/total (the agent cannot set it). A
+        narrative entry without ``plan_step`` is allowed for important
+        mid-step documentation and carries the current derived %.
+        ``percentage`` is only a fallback for tasks with no checklist.
 
         Caller must be the task's assignee and the task must be in an
-        active status — these are the same constraints the pre-gateway
+        active status — same constraints as the pre-gateway handler.
-        `roboco_task_progress` handler enforced.
         """
         active = {
             "in_progress",
@@ -744,16 +751,29 @@ class ContentActions:
                 ),
                 context_briefing={},
             )
-        await self.task.add_progress(
+        result = await self.task.record_plan_progress(
             task_id=task_id,
             agent_id=agent_id,
             message=message,
-            percentage=percentage,
+            plan_step=plan_step,
+            fallback_percentage=percentage,
         )
+        if result is None:
+            return Envelope.not_found(message=f"task {task_id} not found")
+        if result["step_resolved"] is False:
+            valid = result["valid_steps"]
+            return Envelope.invalid_state(
+                message=f"plan_step {plan_step!r} does not match any plan step",
+                remediate=(
+                    "pass a sub_task id or its 1-based order. Valid steps: "
+                    f"{valid}. Re-read them with evidence(task_id)."
+                ),
+                context_briefing={},
+            )
         return Envelope.ok(
             status=str(t.status),
             task_id=str(task_id),
-            next="continue",
+            next=f"progress {result['percentage']}% — continue",
             context_briefing={},
         )
 

roboco/services/task.py

@@ -75,6 +75,46 @@ _MAX_NOTES_CHARS = 8000
 _TRUNCATION_MARKER = "[...earlier notes truncated for size...]\n"
 
 
+def _mark_subtask_complete(sub_tasks: list[dict[str, Any]], plan_step: str) -> bool:
+    """Mark the matching sub_task ``completed`` in place (#173).
+
+    ``plan_step`` matches a sub_task by its id, its ``order``, or its
+    1-based position. Returns True iff a sub_task matched (mutated in
+    place); False when nothing matched.
+    """
+    ref = str(plan_step).strip()
+    for i, st in enumerate(sub_tasks):
+        if ref in {str(st.get("id")), str(st.get("order")), str(i + 1)}:
+            st["completed"] = True
+            return True
+    return False
+
+
+def _plan_subtasks(task: Any) -> tuple[dict[str, Any], list[dict[str, Any]]]:
+    """(plan dict, sub_tasks list) for a task — safe on str/None plans."""
+    plan = task.plan if isinstance(task.plan, dict) else {}
+    sub_tasks = [st for st in (plan.get("sub_tasks") or []) if isinstance(st, dict)]
+    return plan, sub_tasks
+
+
+def _valid_step_refs(sub_tasks: list[dict[str, Any]]) -> list[str]:
+    """Human-listable step refs (id, else order, else 1-based index)."""
+    return [
+        str(st.get("id") or st.get("order") or i + 1) for i, st in enumerate(sub_tasks)
+    ]
+
+
+def _derive_plan_pct(
+    sub_tasks: list[dict[str, Any]], fallback: int | None
+) -> int | None:
+    """% = completed/total of the checklist (equal weight); ``fallback``
+    only when there is no checklist (#173)."""
+    if not sub_tasks:
+        return fallback
+    done = sum(1 for st in sub_tasks if st.get("completed"))
+    return round(done / len(sub_tasks) * 100)
+
+
 def _append_capped(existing: str | None, addition: str) -> str:
     """Append `addition` to `existing`, capped at _MAX_NOTES_CHARS.
 
@@ -3633,6 +3673,61 @@ class TaskService(BaseService):
 
         return task
 
+    async def record_plan_progress(
+        self,
+        task_id: UUID,
+        agent_id: UUID,
+        message: str,
+        plan_step: str | None = None,
+        fallback_percentage: int | None = None,
+    ) -> dict[str, Any] | None:
+        """Append a progress update whose % is DERIVED from the plan checklist.
+
+        #173: the plan's sub_tasks ARE the progress skeleton. When
+        ``plan_step`` (a sub_task id, or 1-based order/index) is given,
+        that step is marked ``completed`` and the percentage is computed
+        as completed/total (equal weight) — the agent cannot game it.
+        Narrative entries (no plan_step) are allowed for documentation
+        and carry the CURRENT derived % so the bar never regresses. When
+        the task has no sub_task checklist the agent's
+        ``fallback_percentage`` is used (back-compat).
+
+        Returns ``None`` if the task is missing, else a dict:
+        ``{"task", "percentage", "step_resolved": bool | None,
+        "valid_steps": [str]}``. ``step_resolved`` is None when no
+        plan_step was requested; False when requested but unmatched (the
+        caller surfaces a remediation listing ``valid_steps``).
+        """
+        task = await self.get(task_id)
+        if not task:
+            return None
+
+        plan, sub_tasks = _plan_subtasks(task)
+        step_resolved: bool | None = None
+        if plan_step is not None:
+            step_resolved = _mark_subtask_complete(sub_tasks, plan_step)
+            if step_resolved:
+                # Reassign so the JSON column registers the mutation.
+                task.plan = {**plan, "sub_tasks": sub_tasks}
+
+        percentage = _derive_plan_pct(sub_tasks, fallback_percentage)
+        task.progress_updates = [
+            *task.progress_updates,
+            {
+                "timestamp": datetime.now(UTC).isoformat(),
+                "agent_id": str(agent_id),
+                "message": message,
+                "percentage": percentage,
+            },
+        ]
+        await self.session.flush()
+        return {
+            "task": task,
+            "percentage": percentage,
+            "step_resolved": step_resolved,
+            "valid_steps": _valid_step_refs(sub_tasks),
+        }
+
     async def add_checkpoint(
         self,
         task_id: UUID,

tests/unit/gateway/test_content_actions.py

@@ -709,3 +709,92 @@ async def test_reflect_incomplete_input_includes_call_example() -> None:
     assert "what_done=" in remediate, remediate
     assert "what_learned=" in remediate, remediate
     assert "what_struggled=" in remediate, remediate
+
+
+# ---------------------------------------------------------------------------
+# #173: plan-driven progress — progress() marks a plan step + derives %.
+# ---------------------------------------------------------------------------
+
+
+def _active_task(agent_id: object) -> MagicMock:
+    return MagicMock(id=uuid4(), assigned_to=agent_id, status="in_progress")
+
+
+@pytest.mark.asyncio
+async def test_progress_plan_step_resolved_ok_with_derived_pct() -> None:
+    agent_id = uuid4()
+    t = _active_task(agent_id)
+    task = AsyncMock()
+    task.get.return_value = t
+    task.record_plan_progress.return_value = {
+        "task": t,
+        "percentage": 50,
+        "step_resolved": True,
+        "valid_steps": ["s1", "s2"],
+    }
+    actions = ContentActions(_make_deps(task=task))
+
+    env = await actions.progress(
+        agent_id=agent_id, task_id=t.id, message="did step 1", plan_step="s1"
+    )
+    body = env.as_dict()
+    assert body.get("error") is None, body
+    assert "50%" in body["next"], body
+    task.record_plan_progress.assert_awaited_once()
+    assert task.record_plan_progress.await_args.kwargs["plan_step"] == "s1"
+
+
+@pytest.mark.asyncio
+async def test_progress_unknown_plan_step_invalid_state_lists_valid() -> None:
+    agent_id = uuid4()
+    t = _active_task(agent_id)
+    task = AsyncMock()
+    task.get.return_value = t
+    task.record_plan_progress.return_value = {
+        "task": t,
+        "percentage": 0,
+        "step_resolved": False,
+        "valid_steps": ["s1", "s2"],
+    }
+    actions = ContentActions(_make_deps(task=task))
+
+    env = await actions.progress(
+        agent_id=agent_id, task_id=t.id, message="?", plan_step="bogus"
+    )
+    body = env.as_dict()
+    assert body["error"] == "invalid_state", body
+    assert "s1" in body["remediate"] and "s2" in body["remediate"], body
+
+
+@pytest.mark.asyncio
+async def test_progress_narrative_without_plan_step_ok() -> None:
+    agent_id = uuid4()
+    t = _active_task(agent_id)
+    task = AsyncMock()
+    task.get.return_value = t
+    task.record_plan_progress.return_value = {
+        "task": t,
+        "percentage": 25,
+        "step_resolved": None,
+        "valid_steps": ["s1"],
+    }
+    actions = ContentActions(_make_deps(task=task))
+
+    env = await actions.progress(
+        agent_id=agent_id, task_id=t.id, message="midway milestone"
+    )
+    assert env.as_dict().get("error") is None
+    assert task.record_plan_progress.await_args.kwargs["plan_step"] is None
+
+
+@pytest.mark.asyncio
+async def test_progress_ownership_enforced() -> None:
+    agent_id = uuid4()
+    t = MagicMock(id=uuid4(), assigned_to=uuid4(), status="in_progress")
+    task = AsyncMock()
+    task.get.return_value = t
+    actions = ContentActions(_make_deps(task=task))
+
+    env = await actions.progress(agent_id=agent_id, task_id=t.id, message="x")
+    assert env.as_dict()["error"] is not None
+    task.record_plan_progress.assert_not_awaited()

tests/unit/services/test_record_plan_progress.py

@@ -0,0 +1,120 @@
+"""#173: progress % is derived from the plan checklist, not agent-set.
+
+The plan's sub_tasks ARE the progress skeleton. progress(plan_step=...)
+marks that step completed and the percentage is computed from
+completed/total (equal weight) — the agent cannot game it. Narrative
+entries (no plan_step) carry the current derived %. Tasks with no
+sub_task checklist fall back to the supplied percentage (back-compat).
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+from uuid import uuid4
+
+import pytest
+from roboco.services.task import TaskService
+
+_PCT_HALF = 50
+_PCT_FULL = 100
+_PCT_NONE = 0
+_PCT_FALLBACK = 42
+
+
+def _svc_with_task(task: Any) -> TaskService:
+    svc = TaskService.__new__(TaskService)
+    svc.get = AsyncMock(return_value=task)  # type: ignore[method-assign]
+    svc.session = MagicMock()
+    svc.session.flush = AsyncMock()
+    return svc
+
+
+def _task_with_plan(sub_tasks: list[dict[str, Any]]) -> MagicMock:
+    return MagicMock(
+        id=uuid4(),
+        plan={"text": "p", "sub_tasks": sub_tasks},
+        progress_updates=[],
+    )
+
+
+@pytest.mark.asyncio
+async def test_marking_step_by_id_derives_percentage() -> None:
+    sid = str(uuid4())
+    task = _task_with_plan(
+        [
+            {"id": sid, "title": "A", "completed": False},
+            {"id": str(uuid4()), "title": "B", "completed": False},
+        ]
+    )
+    svc = _svc_with_task(task)
+    agent = uuid4()
+
+    res = await svc.record_plan_progress(task.id, agent, "did A", plan_step=sid)
+    assert res is not None
+    assert res["step_resolved"] is True
+    assert res["percentage"] == _PCT_HALF  # 1 of 2
+    assert task.plan["sub_tasks"][0]["completed"] is True
+    assert task.progress_updates[-1]["percentage"] == _PCT_HALF
+    assert task.progress_updates[-1]["message"] == "did A"
+
+
+@pytest.mark.asyncio
+async def test_marking_step_by_one_based_order() -> None:
+    task = _task_with_plan(
+        [
+            {"id": "x", "title": "A", "completed": True},
+            {"id": "y", "title": "B", "completed": False},
+        ]
+    )
+    svc = _svc_with_task(task)
+    res = await svc.record_plan_progress(task.id, uuid4(), "did B", plan_step="2")
+    assert res["step_resolved"] is True
+    assert res["percentage"] == _PCT_FULL  # both now complete
+    assert task.plan["sub_tasks"][1]["completed"] is True
+
+
+@pytest.mark.asyncio
+async def test_unknown_step_is_not_resolved_and_lists_valid() -> None:
+    task = _task_with_plan([{"id": "s1", "title": "A", "completed": False}])
+    svc = _svc_with_task(task)
+    res = await svc.record_plan_progress(task.id, uuid4(), "?", plan_step="nope")
+    assert res["step_resolved"] is False
+    assert res["valid_steps"] == ["s1"]
+    # Nothing marked; % still derived from (unchanged) checklist = 0.
+    assert task.plan["sub_tasks"][0]["completed"] is False
+    assert res["percentage"] == _PCT_NONE
+
+
+@pytest.mark.asyncio
+async def test_narrative_entry_carries_current_derived_pct() -> None:
+    task = _task_with_plan(
+        [
+            {"id": "a", "title": "A", "completed": True},
+            {"id": "b", "title": "B", "completed": False},
+        ]
+    )
+    svc = _svc_with_task(task)
+    res = await svc.record_plan_progress(task.id, uuid4(), "midway note")
+    assert res["step_resolved"] is None  # no plan_step requested
+    assert res["percentage"] == _PCT_HALF  # current checklist state
+    assert task.progress_updates[-1]["message"] == "midway note"
+
+
+@pytest.mark.asyncio
+async def test_no_checklist_falls_back_to_supplied_percentage() -> None:
+    task = MagicMock(id=uuid4(), plan="just a string plan", progress_updates=[])
+    svc = _svc_with_task(task)
+    res = await svc.record_plan_progress(
+        task.id, uuid4(), "legacy", fallback_percentage=_PCT_FALLBACK
+    )
+    assert res["percentage"] == _PCT_FALLBACK
+    assert res["valid_steps"] == []
+    assert task.progress_updates[-1]["percentage"] == _PCT_FALLBACK
+
+
+@pytest.mark.asyncio
+async def test_missing_task_returns_none() -> None:
+    svc = TaskService.__new__(TaskService)
+    svc.get = AsyncMock(return_value=None)  # type: ignore[method-assign]
+    assert await svc.record_plan_progress(uuid4(), uuid4(), "x") is None