feat(session): path-key /rewind + /retry into shared verb handlers (#549)

Lift the conversation-modifying /rewind and /retry verbs out of the body-keyed POST /v1/api/command into path-keyed POST /v1/api/workstreams/{ws_id}/rewind ({turns:N}) and /retry, as make_rewind_handler/make_retry_handler in SharedSessionVerbHandlers (template: make_close_handler/make_cancel_handler), wired on both interactive and coordinator kinds. Closes the last unlifted conversation-modifying surface — coordinator workstreams gain rewind/retry where they had none — and removes the surviving exception to the post-#422 path-keyed URL convention.

Handler shape: auth gate (coord -> admin.coordinator via permission_gate; interactive -> conversation.modify via accepted_permissions) -> busy-gate -> session.rewind(n)/retry() -> always emit clear_ui (incl. rewind-to-zero, carries #503) -> audit (conversation.rewind/retry on both kinds). Retry re-dispatch reuses the shared session_worker.send via a per-kind dispatch_retry closure (hard-reject on busy), not a third hand-rolled thread.

The web /command handler now rejects /rewind+/retry with a pointer to the path-keyed endpoint (BREAKING; 1.6.0aN-tolerant); session.handle_command's branches stay for the terminal CLI. auth.py adds the verbs to both write suffix-sets; Python + TS SDKs, OpenAPI (RewindRequest + server/console specs), the /route/ proxy mounts + audit actions, and coordinator_client all gain them.

Interactive frontend (app.js): the 3 /command POST sites + the hand-typed-slash reroute now hit the path-keyed endpoints; the bare .msg.user rewind selector is kept (matches the server's _find_turn_boundaries, which counts system-nudge user turns). The coordinator frontend rewind UX lands in a follow-up commit (browser-verified).

Tests: route-walk mount/order, /route/ audit rows, required_scope, OpenAPI catalog, SDK body-inspection, and HTTP-level handler behavior (busy-gate, turns validation, clear_ui emit, retry dispatch, audit invocation + swallow).

sdk/typescript/src/server.ts

@@ -211,6 +211,24 @@ export class TurnstoneServer extends BaseClient {
     );
   }
 
+  /** Drop the last `turns` conversation turns. Emits a `clear_ui` event. */
+  async rewind(wsId: string, turns: number): Promise<StatusResponse> {
+    return this.request(
+      "POST",
+      `/v1/api/workstreams/${encodeURIComponent(wsId)}/rewind`,
+      { json: { turns } },
+    );
+  }
+
+  /** Drop the last response and re-send the last user message. */
+  async retry(wsId: string): Promise<StatusResponse> {
+    return this.request(
+      "POST",
+      `/v1/api/workstreams/${encodeURIComponent(wsId)}/retry`,
+      { json: {} },
+    );
+  }
+
   // -- Streaming ------------------------------------------------------------
 
   async *streamEvents(wsId: string): AsyncIterableIterator<ServerEvent> {

tests/test_auth.py

@@ -144,6 +144,12 @@ class TestRequiredScope:
     def test_post_close_needs_write(self):
         assert required_scope("POST", "/api/workstreams/abc/close") == "write"
 
+    def test_post_rewind_needs_write(self):
+        assert required_scope("POST", "/api/workstreams/abc/rewind") == "write"
+
+    def test_post_retry_needs_write(self):
+        assert required_scope("POST", "/api/workstreams/abc/retry") == "write"
+
     def test_get_events_per_ws_needs_read(self):
         assert required_scope("GET", "/api/workstreams/abc/events") == "read"
 
@@ -182,6 +188,12 @@ class TestRequiredScope:
     def test_proxy_v1_approve_needs_approve(self):
         assert required_scope("POST", "/node/node-a/v1/api/workstreams/abc/approve") == "approve"
 
+    def test_proxy_v1_rewind_needs_write(self):
+        assert required_scope("POST", "/node/node-a/v1/api/workstreams/abc/rewind") == "write"
+
+    def test_proxy_v1_retry_needs_write(self):
+        assert required_scope("POST", "/node/node-a/v1/api/workstreams/abc/retry") == "write"
+
     def test_proxy_v1_read_endpoint_needs_read(self):
         assert required_scope("GET", "/node/node-a/v1/api/workstreams") == "read"
 

tests/test_coordinator_client.py

@@ -177,6 +177,8 @@ def test_route_map_matches_console_routes():
     assert _ROUTE_PATHS["send"] == "/v1/api/route/workstreams/{ws_id}/send"
     assert _ROUTE_PATHS["approve"] == "/v1/api/route/workstreams/{ws_id}/approve"
     assert _ROUTE_PATHS["cancel"] == "/v1/api/route/workstreams/{ws_id}/cancel"
+    assert _ROUTE_PATHS["rewind"] == "/v1/api/route/workstreams/{ws_id}/rewind"
+    assert _ROUTE_PATHS["retry"] == "/v1/api/route/workstreams/{ws_id}/retry"
     assert _ROUTE_PATHS["close"] == "/v1/api/route/workstreams/{ws_id}/close"
     # ``delete`` keeps the body-keyed shape — it has its own
     # ``route_workstream_delete`` handler instead of going through

tests/test_openapi.py

@@ -31,6 +31,8 @@ class TestServerSpec:
             "/v1/api/workstreams/{ws_id}/send",
             "/v1/api/workstreams/{ws_id}/approve",
             "/v1/api/workstreams/{ws_id}/cancel",
+            "/v1/api/workstreams/{ws_id}/rewind",
+            "/v1/api/workstreams/{ws_id}/retry",
             "/v1/api/workstreams/{ws_id}/close",
             "/v1/api/workstreams/{ws_id}/events",
             "/v1/api/dashboard",
@@ -144,6 +146,8 @@ class TestConsoleSpec:
             "/v1/api/workstreams/{ws_id}/send",
             "/v1/api/workstreams/{ws_id}/approve",
             "/v1/api/workstreams/{ws_id}/cancel",
+            "/v1/api/workstreams/{ws_id}/rewind",
+            "/v1/api/workstreams/{ws_id}/retry",
             "/v1/api/workstreams/{ws_id}/close",
             "/v1/api/workstreams/{ws_id}/events",
             "/v1/api/workstreams/{ws_id}/history",

tests/test_route_proxy_audit.py

@@ -268,6 +268,8 @@ class TestRouteProxyAudit:
             ("/v1/api/route/workstreams/abc123/send", "route.workstream.send"),
             ("/v1/api/route/workstreams/abc123/approve", "route.approve"),
             ("/v1/api/route/workstreams/abc123/cancel", "route.cancel"),
+            ("/v1/api/route/workstreams/abc123/rewind", "route.rewind"),
+            ("/v1/api/route/workstreams/abc123/retry", "route.retry"),
             ("/v1/api/route/command", "route.command"),
             ("/v1/api/route/plan", "route.plan"),
             ("/v1/api/route/workstreams/abc123/close", "route.workstream.close"),

tests/test_sdk_console.py

@@ -29,6 +29,45 @@ def _mock_transport(
     return httpx.MockTransport(handler)
 
 
+# ---------------------------------------------------------------------------
+# Routing proxy — rewind / retry (#549)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.anyio
+async def test_route_rewind_sends_turns_body():
+    """``route_rewind`` forwards ``{"turns": N}`` through the proxy."""
+    captured_path: list[str] = []
+    captured_body: list[dict] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        captured_path.append(request.url.path)
+        captured_body.append(json.loads(request.content) if request.content else {})
+        return _json_response({"status": "ok", "removed": 3})
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
+        client = AsyncTurnstoneConsole(httpx_client=hc)
+        await client.route_rewind("ws1", turns=3)
+        assert captured_path[0] == "/v1/api/route/workstreams/ws1/rewind"
+        assert captured_body[0] == {"turns": 3}
+
+
+@pytest.mark.anyio
+async def test_route_retry_posts_to_path_keyed_endpoint():
+    transport = _mock_transport(
+        {
+            "POST /v1/api/route/workstreams/ws1/retry": _json_response(
+                {"status": "ok", "retried": True}
+            )
+        }
+    )
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
+        client = AsyncTurnstoneConsole(httpx_client=hc)
+        resp = await client.route_retry("ws1")
+        assert resp["status"] == "ok"
+
+
 # ---------------------------------------------------------------------------
 # Cluster overview
 # ---------------------------------------------------------------------------

tests/test_sdk_server.py

@@ -133,6 +133,39 @@ async def test_close_workstream_sends_valid_json_body():
         assert captured["body"] == {"reason": "task complete"}
 
 
+@pytest.mark.anyio
+async def test_rewind_sends_turns_body():
+    """``rewind()`` must transmit ``{"turns": N}`` — the path-keyed
+    rewind handler reads the body via ``read_json_or_400``, so a no-body
+    send would 400. Inspect the body, not just that the path answered
+    (feedback_mock_transport_body_inspection)."""
+    captured: dict = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        captured["path"] = request.url.path
+        captured["body"] = json.loads(request.content) if request.content else None
+        return httpx.Response(200, json={"status": "ok", "removed": 4})
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
+        client = AsyncTurnstoneServer(httpx_client=hc)
+        resp = await client.rewind("ws1", turns=2)
+        assert captured["path"] == "/v1/api/workstreams/ws1/rewind"
+        assert captured["body"] == {"turns": 2}
+        assert resp.status == "ok"
+
+
+@pytest.mark.anyio
+async def test_retry_posts_to_path_keyed_endpoint():
+    transport = _mock_transport(
+        {"POST /v1/api/workstreams/ws1/retry": _json_response({"status": "ok", "retried": True})}
+    )
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
+        client = AsyncTurnstoneServer(httpx_client=hc)
+        resp = await client.retry("ws1")
+        assert resp.status == "ok"
+
+
 # ---------------------------------------------------------------------------
 # Chat interaction
 # ---------------------------------------------------------------------------

tests/test_session_routes.py

@@ -96,6 +96,28 @@ def test_specific_verbs_register_before_bare_detail() -> None:
     assert paths.index("/api/workstreams/{ws_id}/events") < detail_idx
 
 
+def test_rewind_retry_register_before_bare_detail() -> None:
+    """``/rewind`` and ``/retry`` (issue #549) mount as POST verbs
+    before the bare ``{ws_id}`` GET, like the other interaction verbs."""
+    routes: list[Any] = []
+    register_session_routes(
+        routes,
+        prefix="/api/workstreams",
+        handlers=SharedSessionVerbHandlers(
+            detail=_stub,
+            rewind=_stub,
+            retry=_stub,
+        ),
+    )
+    paths = [r.path for r in routes if isinstance(r, Route)]
+    detail_idx = paths.index("/api/workstreams/{ws_id}")
+    assert paths.index("/api/workstreams/{ws_id}/rewind") < detail_idx
+    assert paths.index("/api/workstreams/{ws_id}/retry") < detail_idx
+    by_path = {p: m for p, m in _route_paths(routes)}
+    assert "POST" in by_path["/api/workstreams/{ws_id}/rewind"]
+    assert "POST" in by_path["/api/workstreams/{ws_id}/retry"]
+
+
 def test_attachment_routes_mount_when_quartet_provided() -> None:
     """All four attachment routes mount when ``handlers.attachments``
     is non-``None`` — the type system requires the four-handler

tests/test_workstream_endpoints.py

@@ -30,6 +30,8 @@ from turnstone.core.session_routes import (
     make_detail_handler,
     make_history_handler,
     make_open_handler,
+    make_retry_handler,
+    make_rewind_handler,
 )
 from turnstone.core.storage._sqlite import SQLiteBackend
 from turnstone.core.workstream import WorkstreamKind
@@ -206,6 +208,172 @@ def settings_client(_inject_storage):
     return TestClient(app)
 
 
+# ===========================================================================
+# Rewind / retry (#549 verb lift)
+# ===========================================================================
+
+
+def _rewind_retry_mocks(*, worker_running=False, rewind_return=4, retry_return="hi"):
+    """Mocked ``(manager, session, enqueued-events)`` for the lifted
+    rewind/retry handlers. ``ws._lock`` is a real lock so the handler's
+    busy-gate ``with ws._lock`` works; ``ui._enqueue`` records events."""
+    import threading
+
+    mock_session = MagicMock()
+    mock_session.rewind.return_value = rewind_return
+    mock_session.retry.return_value = retry_return
+    enqueued: list[dict[str, Any]] = []
+    mock_ui = MagicMock()
+    mock_ui._enqueue.side_effect = lambda ev: enqueued.append(ev)
+    mock_ws = MagicMock()
+    mock_ws.session = mock_session
+    mock_ws.ui = mock_ui
+    mock_ws._lock = threading.Lock()
+    mock_ws._worker_running = worker_running
+    mock_mgr = MagicMock()
+    mock_mgr.get.return_value = mock_ws
+    return mock_mgr, mock_session, enqueued
+
+
+def _verb_cfg(mock_mgr: Any) -> SessionEndpointConfig:
+    return SessionEndpointConfig(
+        permission_gate=None,
+        manager_lookup=lambda _r: (mock_mgr, None),
+        tenant_check=None,
+        not_found_label="Workstream not found",
+        audit_action_prefix="workstream",
+    )
+
+
+def _verb_client(route_path: str, handler: Any) -> TestClient:
+    app = Starlette(
+        routes=[Mount("/v1", routes=[Route(route_path, handler, methods=["POST"])])],
+        middleware=[Middleware(_InjectAuthMiddleware)],
+    )
+    return TestClient(app)
+
+
+def test_rewind_returns_removed_and_emits_clear_ui():
+    mock_mgr, mock_session, enqueued = _rewind_retry_mocks(rewind_return=4)
+    handler = make_rewind_handler(_verb_cfg(mock_mgr))
+    client = _verb_client("/api/workstreams/{ws_id}/rewind", handler)
+    resp = client.post("/v1/api/workstreams/ws1/rewind", json={"turns": 2})
+    assert resp.status_code == 200
+    assert resp.json() == {"status": "ok", "removed": 4}
+    mock_session.rewind.assert_called_once_with(2)
+    assert {"type": "clear_ui"} in enqueued
+
+
+def test_rewind_rejects_non_positive_or_non_int_turns():
+    mock_mgr, mock_session, _ = _rewind_retry_mocks()
+    handler = make_rewind_handler(_verb_cfg(mock_mgr))
+    client = _verb_client("/api/workstreams/{ws_id}/rewind", handler)
+    # ``True`` is an int subclass — must be rejected too.
+    for bad in ({}, {"turns": 0}, {"turns": -1}, {"turns": "two"}, {"turns": True}):
+        resp = client.post("/v1/api/workstreams/ws1/rewind", json=bad)
+        assert resp.status_code == 400, bad
+    mock_session.rewind.assert_not_called()
+
+
+def test_rewind_while_busy_returns_busy_and_skips_mutation():
+    mock_mgr, mock_session, enqueued = _rewind_retry_mocks(worker_running=True)
+    handler = make_rewind_handler(_verb_cfg(mock_mgr))
+    client = _verb_client("/api/workstreams/{ws_id}/rewind", handler)
+    resp = client.post("/v1/api/workstreams/ws1/rewind", json={"turns": 1})
+    assert resp.status_code == 200
+    assert resp.json()["status"] == "busy"
+    mock_session.rewind.assert_not_called()
+    assert any(e.get("type") == "busy_error" for e in enqueued)
+
+
+def test_retry_dispatches_and_emits_clear_ui():
+    mock_mgr, _session, enqueued = _rewind_retry_mocks(retry_return="hello")
+    dispatched: list[str] = []
+    handler = make_retry_handler(
+        _verb_cfg(mock_mgr), dispatch_retry=lambda _ws, msg: dispatched.append(msg)
+    )
+    client = _verb_client("/api/workstreams/{ws_id}/retry", handler)
+    resp = client.post("/v1/api/workstreams/ws1/retry")
+    assert resp.status_code == 200
+    assert resp.json() == {"status": "ok", "retried": True}
+    assert dispatched == ["hello"]
+    assert {"type": "clear_ui"} in enqueued
+
+
+def test_retry_nothing_to_retry_skips_dispatch():
+    mock_mgr, _session, enqueued = _rewind_retry_mocks(retry_return=None)
+    dispatched: list[str] = []
+    handler = make_retry_handler(
+        _verb_cfg(mock_mgr), dispatch_retry=lambda _ws, msg: dispatched.append(msg)
+    )
+    client = _verb_client("/api/workstreams/{ws_id}/retry", handler)
+    resp = client.post("/v1/api/workstreams/ws1/retry")
+    assert resp.status_code == 200
+    assert resp.json() == {"status": "ok", "retried": False}
+    assert dispatched == []
+    assert {"type": "clear_ui"} in enqueued
+
+
+def test_retry_while_busy_returns_busy_and_skips_dispatch():
+    mock_mgr, mock_session, enqueued = _rewind_retry_mocks(worker_running=True)
+    dispatched: list[str] = []
+    handler = make_retry_handler(
+        _verb_cfg(mock_mgr), dispatch_retry=lambda _ws, msg: dispatched.append(msg)
+    )
+    client = _verb_client("/api/workstreams/{ws_id}/retry", handler)
+    resp = client.post("/v1/api/workstreams/ws1/retry")
+    assert resp.status_code == 200
+    assert resp.json()["status"] == "busy"
+    mock_session.retry.assert_not_called()
+    assert dispatched == []
+    assert any(e.get("type") == "busy_error" for e in enqueued)
+
+
+def test_rewind_invokes_audit_emit_with_turns():
+    """The handler calls ``audit_emit(request, ws_id, ws, turns)`` — a
+    dropped ``audit_emit=`` wiring or a renamed arg would break this."""
+    mock_mgr, _session, _enqueued = _rewind_retry_mocks()
+    captured: list[tuple[str, int]] = []
+    handler = make_rewind_handler(
+        _verb_cfg(mock_mgr),
+        audit_emit=lambda _req, ws_id, _ws, turns: captured.append((ws_id, turns)),
+    )
+    client = _verb_client("/api/workstreams/{ws_id}/rewind", handler)
+    resp = client.post("/v1/api/workstreams/ws1/rewind", json={"turns": 3})
+    assert resp.status_code == 200
+    assert captured == [("ws1", 3)]
+
+
+def test_retry_invokes_audit_emit():
+    mock_mgr, _session, _enqueued = _rewind_retry_mocks(retry_return="hi")
+    captured: list[str] = []
+    handler = make_retry_handler(
+        _verb_cfg(mock_mgr),
+        dispatch_retry=lambda _ws, _msg: None,
+        audit_emit=lambda _req, ws_id, _ws: captured.append(ws_id),
+    )
+    client = _verb_client("/api/workstreams/{ws_id}/retry", handler)
+    resp = client.post("/v1/api/workstreams/ws1/retry")
+    assert resp.status_code == 200
+    assert captured == ["ws1"]
+
+
+def test_rewind_swallows_audit_emit_exception():
+    """A raising ``audit_emit`` is demoted to a warning — the handler still
+    returns 200 and the rewind still took effect (mirrors close/cancel)."""
+    mock_mgr, mock_session, _enqueued = _rewind_retry_mocks(rewind_return=2)
+
+    def _boom(_req, _ws_id, _ws, _turns):
+        raise RuntimeError("audit backend down")
+
+    handler = make_rewind_handler(_verb_cfg(mock_mgr), audit_emit=_boom)
+    client = _verb_client("/api/workstreams/{ws_id}/rewind", handler)
+    resp = client.post("/v1/api/workstreams/ws1/rewind", json={"turns": 1})
+    assert resp.status_code == 200
+    assert resp.json() == {"status": "ok", "removed": 2}
+    mock_session.rewind.assert_called_once_with(1)
+
+
 # ===========================================================================
 # DELETE workstream
 # ===========================================================================

turnstone/api/console_spec.py

@@ -136,6 +136,7 @@ from turnstone.api.server_schemas import (
     ListAttachmentsResponse,
     ListSkillSummaryResponse,
     ListWorkstreamsResponse,
+    RewindRequest,
     SkillSummary,
     UploadAttachmentResponse,
     WorkstreamDetailResponse,
@@ -1337,6 +1338,33 @@ CONSOLE_ENDPOINTS: list[EndpointSpec] = [
         error_codes=[403, 404, 503],
         tags=["Coordinator"],
     ),
+    EndpointSpec(
+        "/v1/api/workstreams/{ws_id}/rewind",
+        "POST",
+        "Drop the last N conversation turns on the coordinator (emits clear_ui)",
+        description=(
+            "Truncates the coordinator conversation by N turns via the shared "
+            "rewind handler and emits ``clear_ui`` so the dashboard re-fetches "
+            "the truncated history. Gated on ``admin.coordinator``."
+        ),
+        request_model=RewindRequest,
+        response_model=StatusResponse,
+        error_codes=[400, 403, 404, 503],
+        tags=["Coordinator"],
+    ),
+    EndpointSpec(
+        "/v1/api/workstreams/{ws_id}/retry",
+        "POST",
+        "Re-send the last user message on the coordinator for a fresh response",
+        description=(
+            "Drops the last response and re-sends the last user message via the "
+            "shared worker dispatch, emitting ``clear_ui``. Gated on "
+            "``admin.coordinator``."
+        ),
+        response_model=StatusResponse,
+        error_codes=[400, 403, 404, 503],
+        tags=["Coordinator"],
+    ),
     EndpointSpec(
         "/v1/api/workstreams/{ws_id}/close",
         "POST",

turnstone/api/server_schemas.py

@@ -113,6 +113,14 @@ class CancelRequest(BaseModel):
     )
 
 
+class RewindRequest(BaseModel):
+    turns: int = Field(
+        description="Number of conversation turns (user message + its responses) "
+        "to drop from the end. Clamped to the available turn count.",
+        ge=1,
+    )
+
+
 class CreateWorkstreamRequest(BaseModel):
     name: str = Field(default="", description="Workstream display name (auto-generated if empty)")
     model: str = Field(default="", description="Model alias from registry")

turnstone/api/server_spec.py

@@ -37,6 +37,7 @@ from turnstone.api.server_schemas import (
     ListWorkstreamsResponse,
     MemoryInfo,
     PlanFeedbackRequest,
+    RewindRequest,
     SaveMemoryRequest,
     SearchMemoriesRequest,
     SendRequest,
@@ -151,6 +152,23 @@ SERVER_ENDPOINTS: list[EndpointSpec] = [
         error_codes=[400, 404],
         tags=["Chat"],
     ),
+    EndpointSpec(
+        "/v1/api/workstreams/{ws_id}/rewind",
+        "POST",
+        "Drop the last N conversation turns (emits clear_ui)",
+        request_model=RewindRequest,
+        response_model=StatusResponse,
+        error_codes=[400, 404],
+        tags=["Chat"],
+    ),
+    EndpointSpec(
+        "/v1/api/workstreams/{ws_id}/retry",
+        "POST",
+        "Drop the last response and re-send the last user message",
+        response_model=StatusResponse,
+        error_codes=[400, 404],
+        tags=["Chat"],
+    ),
     # --- Streaming ---
     EndpointSpec(
         "/v1/api/workstreams/{ws_id}/events",
@@ -450,6 +468,7 @@ _ALL_MODELS: list[type[BaseModel]] = [
     PlanFeedbackRequest,
     CommandRequest,
     CancelRequest,
+    RewindRequest,
     CreateWorkstreamRequest,
     CreateWorkstreamResponse,
     CloseWorkstreamRequest,

turnstone/console/coordinator_client.py

@@ -293,6 +293,8 @@ _ROUTE_PATHS: dict[str, str] = {
     "send": "/v1/api/route/workstreams/{ws_id}/send",
     "approve": "/v1/api/route/workstreams/{ws_id}/approve",
     "cancel": "/v1/api/route/workstreams/{ws_id}/cancel",
+    "rewind": "/v1/api/route/workstreams/{ws_id}/rewind",
+    "retry": "/v1/api/route/workstreams/{ws_id}/retry",
     "close": "/v1/api/route/workstreams/{ws_id}/close",
     # ``delete`` is the only surviving body-keyed routing proxy path
     # — it has its own ``route_workstream_delete`` handler instead of
@@ -589,6 +591,16 @@ class CoordinatorClient:
             return {"error": f"workstream not in coordinator subtree: {ws_id}", "status": 404}
         return self._post("cancel", {}, ws_id=ws_id)
 
+    def rewind(self, ws_id: str, turns: int) -> dict[str, Any]:
+        if not self._is_own_subtree(ws_id):
+            return {"error": f"workstream not in coordinator subtree: {ws_id}", "status": 404}
+        return self._post("rewind", {"turns": turns}, ws_id=ws_id)
+
+    def retry(self, ws_id: str) -> dict[str, Any]:
+        if not self._is_own_subtree(ws_id):
+            return {"error": f"workstream not in coordinator subtree: {ws_id}", "status": 404}
+        return self._post("retry", {}, ws_id=ws_id)
+
     # -- model-invoked block-wait -----------------------------------------
 
     # ClassVar aliases for the module-level wait_for_workstream constants.

turnstone/console/server.py

@@ -72,6 +72,8 @@ from turnstone.core.session_routes import (
     make_history_handler,
     make_list_handler,
     make_open_handler,
+    make_retry_handler,
+    make_rewind_handler,
     make_saved_handler,
     make_send_handler,
     register_coord_verbs,
@@ -689,6 +691,8 @@ _ROUTE_PROXY_AUDIT_ACTIONS: dict[str, str] = {
     "dequeue": "route.workstream.dequeue",
     "approve": "route.approve",
     "cancel": "route.cancel",
+    "rewind": "route.rewind",
+    "retry": "route.retry",
     "command": "route.command",
     "plan": "route.plan",
     "close": "route.workstream.close",
@@ -3158,6 +3162,86 @@ def _audit_cancel_coordinator(
     )
 
 
+def _audit_rewind_coordinator(
+    request: Request,
+    ws_id: str,
+    ws_before: Workstream,  # noqa: ARG001 — coord audit detail keys off ws_id
+    turns: int,
+) -> None:
+    """Record the ``conversation.rewind`` audit event for coord rewind.
+
+    Passed to :func:`make_rewind_handler` as ``audit_emit``. The action is
+    hardcoded ``conversation.rewind`` (matching interactive — there is
+    deliberately no ``coordinator.rewind`` split).
+    """
+    storage = getattr(request.app.state, "auth_storage", None)
+    if storage is None:
+        return
+    record_audit(
+        storage,
+        _auth_user_id(request),
+        "conversation.rewind",
+        "workstream",
+        ws_id,
+        {"coord_ws_id": ws_id, "src": "coordinator", "turns": turns},
+        request.client.host if request.client else "",
+    )
+
+
+def _audit_retry_coordinator(
+    request: Request,
+    ws_id: str,
+    ws_before: Workstream,  # noqa: ARG001 — coord audit detail keys off ws_id
+) -> None:
+    """Record the ``conversation.retry`` audit event for coord retry."""
+    storage = getattr(request.app.state, "auth_storage", None)
+    if storage is None:
+        return
+    record_audit(
+        storage,
+        _auth_user_id(request),
+        "conversation.retry",
+        "workstream",
+        ws_id,
+        {"coord_ws_id": ws_id, "src": "coordinator"},
+        request.client.host if request.client else "",
+    )
+
+
+def _coord_dispatch_retry(ws: Workstream, user_msg: str) -> None:
+    """Re-send ``user_msg`` on a coordinator workstream after ``/retry``.
+
+    Passed to :func:`make_retry_handler` as ``dispatch_retry``. Mirrors
+    :meth:`CoordinatorAdapter.send`'s worker shape — drives the shared
+    :func:`turnstone.core.session_worker.send` dispatcher — but without
+    attachment handling (a retry re-sends an existing text turn). The
+    ``run`` closure's error handling is intentionally light:
+    :meth:`ChatSession.send` already surfaces failures to SSE, persists
+    ``last_error`` and emits state=error via ``_record_fatal_error``, and
+    the shared dispatcher owns the ``_worker_running`` lifecycle — so the
+    worker only logs. The ``enqueue`` closure hard-rejects (a retry must
+    not silently queue behind an in-flight turn).
+    """
+    from turnstone.core import session_worker
+
+    session = ws.session
+    ui = ws.ui
+    if session is None:
+        return
+
+    def _run() -> None:
+        try:
+            session.send(user_msg)
+        except Exception:
+            log.exception("coord.retry.worker_failed ws=%s", ws.id[:8])
+
+    def _enqueue() -> None:
+        if ui is not None and hasattr(ui, "on_error"):
+            ui.on_error("Cannot retry: workstream is busy")
+
+    session_worker.send(ws, enqueue=_enqueue, run=_run, thread_name=f"coord-retry-{ws.id[:8]}")
+
+
 def _coord_events_replay(
     ws: Workstream,
     ui: Any,
@@ -12565,6 +12649,15 @@ def create_app(
                 coord_endpoint_config,
                 audit_emit=_audit_cancel_coordinator,
             ),
+            rewind=make_rewind_handler(  # lifted: shared body (#549)
+                coord_endpoint_config,
+                audit_emit=_audit_rewind_coordinator,
+            ),
+            retry=make_retry_handler(  # lifted: shared body (#549)
+                coord_endpoint_config,
+                dispatch_retry=_coord_dispatch_retry,
+                audit_emit=_audit_retry_coordinator,
+            ),
             events=make_events_handler(coord_endpoint_config),  # lifted: shared body
             history=make_history_handler(coord_endpoint_config),  # lifted: shared body
             attachments=make_attachment_handlers(
@@ -12619,6 +12712,16 @@ def create_app(
                         route_proxy,
                         methods=["POST"],
                     ),
+                    Route(
+                        "/api/route/workstreams/{ws_id}/rewind",
+                        route_proxy,
+                        methods=["POST"],
+                    ),
+                    Route(
+                        "/api/route/workstreams/{ws_id}/retry",
+                        route_proxy,
+                        methods=["POST"],
+                    ),
                     Route("/api/route/command", route_proxy, methods=["POST"]),
                     Route("/api/route/plan", route_proxy, methods=["POST"]),
                     Route(

turnstone/core/auth.py

@@ -696,6 +696,8 @@ def required_scope(method: str, path: str) -> str:
             "attachments",
             "send",
             "cancel",
+            "rewind",
+            "retry",
             "close",
         }
     ):
@@ -744,6 +746,8 @@ def required_scope(method: str, path: str) -> str:
                 "attachments",
                 "send",
                 "cancel",
+                "rewind",
+                "retry",
                 "close",
             }:
                 return "write"

turnstone/core/session_routes.py

@@ -518,6 +518,8 @@ class SharedSessionVerbHandlers:
     approve: Handler | None = None  # POST {prefix}/{ws_id}/approve
     plan: Handler | None = None  # POST {prefix}/{ws_id}/plan
     cancel: Handler | None = None  # POST {prefix}/{ws_id}/cancel
+    rewind: Handler | None = None  # POST {prefix}/{ws_id}/rewind
+    retry: Handler | None = None  # POST {prefix}/{ws_id}/retry
     events: Handler | None = None  # GET  {prefix}/{ws_id}/events (SSE)
     history: Handler | None = None  # GET  {prefix}/{ws_id}/history
 
@@ -605,6 +607,10 @@ def register_session_routes(
         routes.append(Route(f"{p}/{{ws_id}}/plan", handlers.plan, methods=["POST"]))
     if handlers.cancel is not None:
         routes.append(Route(f"{p}/{{ws_id}}/cancel", handlers.cancel, methods=["POST"]))
+    if handlers.rewind is not None:
+        routes.append(Route(f"{p}/{{ws_id}}/rewind", handlers.rewind, methods=["POST"]))
+    if handlers.retry is not None:
+        routes.append(Route(f"{p}/{{ws_id}}/retry", handlers.retry, methods=["POST"]))
     if handlers.events is not None:
         routes.append(Route(f"{p}/{{ws_id}}/events", handlers.events, methods=["GET"]))
     if handlers.history is not None:
@@ -1187,6 +1193,241 @@ def make_cancel_handler(
     return cancel
 
 
+RewindAuditEmitter = Callable[
+    ["Request", str, "Workstream", int],
+    None,
+]
+RetryAuditEmitter = Callable[
+    ["Request", str, "Workstream"],
+    None,
+]
+# (ws, user_msg) -> None. Re-sends ``user_msg`` on ``ws`` via the kind's
+# worker dispatch (driving :func:`turnstone.core.session_worker.send`
+# with the kind's own run / enqueue callbacks). The retry handler calls
+# it after :meth:`ChatSession.retry` truncates the last turn.
+RetryDispatcher = Callable[
+    ["Workstream", str],
+    None,
+]
+
+
+def make_rewind_handler(
+    cfg: SessionEndpointConfig,
+    *,
+    audit_emit: RewindAuditEmitter | None = None,
+    accepted_permissions: tuple[str, ...] = (),
+) -> Handler:
+    """Lifted body for ``POST {prefix}/{ws_id}/rewind`` (body ``{"turns": N}``).
+
+    Drops the last ``N`` conversation turns via :meth:`ChatSession.rewind`
+    (kind-agnostic: mutates ``messages`` + ``_msg_tokens`` + storage, with
+    attachment / FTS cleanup riding the app-level cascade inside
+    ``delete_messages_after`` — which is exactly why both kinds reuse
+    ``rewind()`` rather than bespoke SQL). Both kinds share the auth →
+    mgr → ws-lookup → busy-gate → ``rewind`` → ``clear_ui`` → audit
+    sequence.
+
+    Unlike :func:`make_close_handler`, the body emits a ``clear_ui`` event
+    after the mutation — **always, including a rewind to zero messages**.
+    The frontend keys its REST ``/history`` refetch (and any queued
+    edit-and-resend) off this signal, not an inline history payload; the
+    unconditional emit carries the PR #503 fix (an ``if history:`` guard
+    once froze the composer on rewind-to-zero).
+
+    Args:
+        cfg: per-kind policy bundle (auth, manager lookup, tenant check,
+            error labels). Captured by closure.
+        audit_emit: kind's audit emitter; receives
+            ``(request, ws_id, ws, turns)``. Wrapped in try/except — an
+            audit-write failure logs a warning, never an HTTP 500.
+            **Both kinds hardcode the ``conversation.rewind`` action**
+            (NOT ``cfg.audit_action_prefix`` — there is deliberately no
+            ``coordinator.rewind`` split).
+        accepted_permissions: fallback scope check used only when
+            ``cfg.permission_gate`` is ``None``. Interactive wires
+            ``("conversation.modify",)``; coord leaves it empty and
+            relies on its ``admin.coordinator`` ``permission_gate``.
+    """
+
+    async def rewind(request: Request) -> Response:
+        import asyncio
+
+        from turnstone.core.auth import require_any_permission
+        from turnstone.core.web_helpers import read_json_or_400
+
+        if cfg.permission_gate is not None:
+            err = cfg.permission_gate(request)
+            if err is not None:
+                return err
+        elif accepted_permissions:
+            err = require_any_permission(request, accepted_permissions)
+            if err is not None:
+                return err
+        mgr_opt, err503 = cfg.manager_lookup(request)
+        if err503 is not None:
+            return err503
+        # See ``make_approve_handler`` for the cast rationale.
+        mgr = cast("SessionManager", mgr_opt)
+        ws_id = request.path_params.get("ws_id", "")
+
+        body = await read_json_or_400(request)
+        if isinstance(body, JSONResponse):
+            return body
+        raw_turns = body.get("turns")
+        # ``bool`` is an ``int`` subclass — reject it explicitly so
+        # ``{"turns": true}`` can't sneak through as "rewind 1".
+        if not isinstance(raw_turns, int) or isinstance(raw_turns, bool) or raw_turns < 1:
+            return JSONResponse(
+                {"error": "turns must be a positive integer"},
+                status_code=400,
+            )
+
+        if cfg.tenant_check is not None:
+            err_tenant = await asyncio.to_thread(cfg.tenant_check, request, ws_id, mgr)
+            if err_tenant is not None:
+                return err_tenant
+
+        ws = mgr.get(ws_id)
+        if ws is None:
+            return JSONResponse({"error": cfg.not_found_label}, status_code=404)
+        session = ws.session
+        ui = ws.ui
+        if session is None or ui is None:
+            return JSONResponse({"error": "No session"}, status_code=400)
+
+        # Reject rewind while a generation is in flight — mutating
+        # ``messages`` under a running worker corrupts history / cursors.
+        # Gate on ``_worker_running`` (not ``worker_thread.is_alive()``)
+        # for parity with session_worker.send.
+        with ws._lock:
+            if ws._worker_running:
+                if hasattr(ui, "_enqueue"):
+                    ui._enqueue(
+                        {"type": "busy_error", "message": "Cannot rewind while processing."}
+                    )
+                return JSONResponse({"status": "busy"})
+
+        removed = session.rewind(raw_turns)
+
+        if hasattr(ui, "_enqueue"):
+            ui._enqueue({"type": "clear_ui"})
+
+        if audit_emit is not None:
+            try:
+                audit_emit(request, ws_id, ws, raw_turns)
+            except Exception:
+                log.warning(
+                    "ws.rewind.audit_failed ws=%s",
+                    ws_id[:8] if ws_id else "",
+                    exc_info=True,
+                )
+
+        return JSONResponse({"status": "ok", "removed": removed})
+
+    return rewind
+
+
+def make_retry_handler(
+    cfg: SessionEndpointConfig,
+    *,
+    dispatch_retry: RetryDispatcher,
+    audit_emit: RetryAuditEmitter | None = None,
+    accepted_permissions: tuple[str, ...] = (),
+) -> Handler:
+    """Lifted body for ``POST {prefix}/{ws_id}/retry`` (no body).
+
+    Drops the last assistant response via :meth:`ChatSession.retry` and
+    re-sends the last user message for a fresh generation. Shares the
+    auth → mgr → ws-lookup → busy-gate → ``retry`` → ``clear_ui`` →
+    audit → re-dispatch sequence across kinds.
+
+    The re-send goes through ``dispatch_retry`` (a per-kind closure that
+    drives :func:`turnstone.core.session_worker.send` with the kind's own
+    ``run`` / ``enqueue`` callbacks) rather than a hand-rolled thread, so
+    both kinds converge on the shared worker-dispatch primitive instead
+    of open-coding a third copy. A retry issued while busy is rejected up
+    front by the busy-gate below; the dispatcher's ``enqueue`` callback
+    hard-rejects (rather than queues) so the rare check-then-dispatch
+    race can't silently defer the resend behind the in-flight turn.
+
+    ``clear_ui`` fires after ``retry()`` regardless of whether anything
+    was dropped (idempotent REST refetch on the frontend), matching the
+    pre-lift interactive handler.
+
+    Args:
+        cfg: per-kind policy bundle.
+        dispatch_retry: ``(ws, user_msg) -> None`` re-send closure.
+            Required — retry is meaningless without it.
+        audit_emit: kind's audit emitter; receives ``(request, ws_id,
+            ws)``. **Both kinds hardcode the ``conversation.retry``
+            action.** Wrapped in try/except.
+        accepted_permissions: fallback scope check used only when
+            ``cfg.permission_gate`` is ``None`` (interactive wires
+            ``("conversation.modify",)``).
+    """
+
+    async def retry(request: Request) -> Response:
+        import asyncio
+
+        from turnstone.core.auth import require_any_permission
+
+        if cfg.permission_gate is not None:
+            err = cfg.permission_gate(request)
+            if err is not None:
+                return err
+        elif accepted_permissions:
+            err = require_any_permission(request, accepted_permissions)
+            if err is not None:
+                return err
+        mgr_opt, err503 = cfg.manager_lookup(request)
+        if err503 is not None:
+            return err503
+        mgr = cast("SessionManager", mgr_opt)
+        ws_id = request.path_params.get("ws_id", "")
+
+        if cfg.tenant_check is not None:
+            err_tenant = await asyncio.to_thread(cfg.tenant_check, request, ws_id, mgr)
+            if err_tenant is not None:
+                return err_tenant
+
+        ws = mgr.get(ws_id)
+        if ws is None:
+            return JSONResponse({"error": cfg.not_found_label}, status_code=404)
+        session = ws.session
+        ui = ws.ui
+        if session is None or ui is None:
+            return JSONResponse({"error": "No session"}, status_code=400)
+
+        with ws._lock:
+            if ws._worker_running:
+                if hasattr(ui, "_enqueue"):
+                    ui._enqueue({"type": "busy_error", "message": "Cannot retry while processing."})
+                return JSONResponse({"status": "busy"})
+
+        retry_msg = session.retry()
+
+        if hasattr(ui, "_enqueue"):
+            ui._enqueue({"type": "clear_ui"})
+
+        if audit_emit is not None:
+            try:
+                audit_emit(request, ws_id, ws)
+            except Exception:
+                log.warning(
+                    "ws.retry.audit_failed ws=%s",
+                    ws_id[:8] if ws_id else "",
+                    exc_info=True,
+                )
+
+        retried = retry_msg is not None
+        if retry_msg is not None:
+            dispatch_retry(ws, retry_msg)
+
+        return JSONResponse({"status": "ok", "retried": retried})
+
+    return retry
+
+
 def make_open_handler(
     cfg: SessionEndpointConfig,
     *,

turnstone/sdk/console.py

@@ -445,6 +445,16 @@ class AsyncTurnstoneConsole(_BaseClient):
             "POST", f"/v1/api/route/workstreams/{ws_id}/cancel", json_body=body
         )
 
+    async def route_rewind(self, ws_id: str, *, turns: int) -> dict[str, Any]:
+        """Drop the last ``turns`` conversation turns via the routing proxy."""
+        return await self._request(
+            "POST", f"/v1/api/route/workstreams/{ws_id}/rewind", json_body={"turns": turns}
+        )
+
+    async def route_retry(self, ws_id: str) -> dict[str, Any]:
+        """Re-send the last user message for a fresh response via the routing proxy."""
+        return await self._request("POST", f"/v1/api/route/workstreams/{ws_id}/retry", json_body={})
+
     async def route_command(self, *, ws_id: str, command: str) -> dict[str, Any]:
         """Send a slash command via the routing proxy."""
         return await self._request(
@@ -1359,6 +1369,12 @@ class TurnstoneConsole:
     def route_cancel(self, ws_id: str, *, force: bool = False) -> dict[str, Any]:
         return self._runner.run(self._async.route_cancel(ws_id, force=force))
 
+    def route_rewind(self, ws_id: str, *, turns: int) -> dict[str, Any]:
+        return self._runner.run(self._async.route_rewind(ws_id, turns=turns))
+
+    def route_retry(self, ws_id: str) -> dict[str, Any]:
+        return self._runner.run(self._async.route_retry(ws_id))
+
     def route_command(self, *, ws_id: str, command: str) -> dict[str, Any]:
         return self._runner.run(self._async.route_command(ws_id=ws_id, command=command))
 

turnstone/sdk/server.py

@@ -310,6 +310,24 @@ class AsyncTurnstoneServer(_BaseClient):
             response_model=StatusResponse,
         )
 
+    async def rewind(self, ws_id: str, *, turns: int) -> StatusResponse:
+        """Drop the last ``turns`` conversation turns. Emits ``clear_ui``."""
+        return await self._request(
+            "POST",
+            f"/v1/api/workstreams/{ws_id}/rewind",
+            json_body={"turns": turns},
+            response_model=StatusResponse,
+        )
+
+    async def retry(self, ws_id: str) -> StatusResponse:
+        """Drop the last response and re-send the last user message."""
+        return await self._request(
+            "POST",
+            f"/v1/api/workstreams/{ws_id}/retry",
+            json_body={},
+            response_model=StatusResponse,
+        )
+
     # -- streaming -----------------------------------------------------------
 
     async def stream_events(self, ws_id: str) -> AsyncIterator[ServerEvent]:
@@ -680,6 +698,12 @@ class TurnstoneServer:
     def cancel(self, ws_id: str, *, force: bool = False) -> StatusResponse:
         return self._runner.run(self._async.cancel(ws_id, force=force))
 
+    def rewind(self, ws_id: str, *, turns: int) -> StatusResponse:
+        return self._runner.run(self._async.rewind(ws_id, turns=turns))
+
+    def retry(self, ws_id: str) -> StatusResponse:
+        return self._runner.run(self._async.retry(ws_id))
+
     # -- streaming -----------------------------------------------------------
 
     def stream_events(self, ws_id: str) -> Iterator[ServerEvent]:

turnstone/server.py

@@ -75,6 +75,8 @@ from turnstone.core.session_routes import (
     make_history_handler,
     make_list_handler,
     make_open_handler,
+    make_retry_handler,
+    make_rewind_handler,
     make_saved_handler,
     make_send_handler,
     register_session_routes,
@@ -609,6 +611,98 @@ def _audit_close_workstream(
     )
 
 
+def _audit_rewind_workstream(
+    request: Request,
+    ws_id: str,
+    ws_before: Workstream,  # noqa: ARG001 — detail keys off ws_id
+    turns: int,
+) -> None:
+    """Record the ``conversation.rewind`` audit event for interactive rewind.
+
+    Passed to :func:`make_rewind_handler` as ``audit_emit``. The action
+    is hardcoded ``conversation.rewind`` on both kinds (no
+    ``coordinator.rewind`` split).
+    """
+    from turnstone.core.audit import record_audit
+
+    storage = getattr(request.app.state, "auth_storage", None)
+    if storage is None:
+        return
+    _, ip = _audit_context(request)
+    record_audit(
+        storage,
+        _auth_user_id(request),
+        "conversation.rewind",
+        "workstream",
+        ws_id,
+        {"turns": turns, "ws_id": ws_id},
+        ip,
+    )
+
+
+def _audit_retry_workstream(
+    request: Request,
+    ws_id: str,
+    ws_before: Workstream,  # noqa: ARG001 — detail keys off ws_id
+) -> None:
+    """Record the ``conversation.retry`` audit event for interactive retry."""
+    from turnstone.core.audit import record_audit
+
+    storage = getattr(request.app.state, "auth_storage", None)
+    if storage is None:
+        return
+    _, ip = _audit_context(request)
+    record_audit(
+        storage,
+        _auth_user_id(request),
+        "conversation.retry",
+        "workstream",
+        ws_id,
+        {"ws_id": ws_id},
+        ip,
+    )
+
+
+def _interactive_dispatch_retry(ws: Workstream, user_msg: str) -> None:
+    """Re-send ``user_msg`` on an interactive workstream after ``/retry``.
+
+    Passed to :func:`make_retry_handler` as ``dispatch_retry``; called
+    once :meth:`ChatSession.retry` has truncated the last turn. Drives
+    the shared :func:`turnstone.core.session_worker.send` dispatcher with
+    an interactive ``run`` closure (surfaces ``GenerationCancelled`` /
+    errors through the WebUI hooks) and a hard-reject ``enqueue`` closure
+    (a retry must not silently queue behind an in-flight turn — preserves
+    the pre-lift inline behaviour). The shared dispatcher owns the
+    ``_worker_running`` lifecycle, so the ``run`` closure needs no
+    ``finally`` flag-clear of its own.
+    """
+    from turnstone.core import session_worker
+
+    session = ws.session
+    ui = ws.ui
+    if session is None or ui is None:
+        return
+
+    def _run() -> None:
+        me = threading.current_thread()
+        try:
+            session.send(user_msg)
+        except GenerationCancelled:
+            if ws.worker_thread is me:
+                ui.on_stream_end()
+                ui.on_state_change("idle")
+        except Exception as exc:
+            if ws.worker_thread is me:
+                ui.on_error(f"Error: {exc}")
+                ui.on_stream_end()
+                ui.on_state_change("error")
+
+    def _enqueue() -> None:
+        ui.on_error("Cannot retry: workstream is busy")
+
+    session_worker.send(ws, enqueue=_enqueue, run=_run, thread_name=f"retry-{ws.id[:8]}")
+
+
 def _interactive_events_replay(
     ws: Workstream, ui: Any, request: Request
 ) -> Iterable[dict[str, Any]]:
@@ -1286,28 +1380,24 @@ async def command(request: Request) -> JSONResponse:
     try:
         # Permission gate for conversation-modifying commands
         cmd_word = cmd.strip().split(None, 1)[0].lower()
+        # ``/rewind`` and ``/retry`` were lifted to path-keyed endpoints
+        # (POST /v1/api/workstreams/{ws_id}/rewind|retry, issue #549).
+        # Reject them here so a stale web client gets a clear pointer
+        # instead of a half-applied mutation: ``handle_command`` below
+        # would still rewind/retry, but the web-specific clear_ui emit +
+        # retry re-dispatch no longer live in this handler. The terminal
+        # CLI keeps dispatching these through ``handle_command`` in-process.
         if cmd_word in ("/rewind", "/retry"):
-            from turnstone.core.auth import require_permission
-
-            err = require_permission(request, "conversation.modify")
-            if err:
-                ui.on_error("Permission denied: conversation.modify required")
-                return err
-            # Prevent rewind/retry while a generation is in progress.
-            # Gate on ``_worker_running`` (not ``worker_thread.is_alive()``)
-            # for parity with session_worker.send: spawn paths set the
-            # flag before assigning ws.worker_thread, so a reader using
-            # the old gate could see a stale dead thread while a new
-            # worker is in the middle of starting.
-            with ws._lock:
-                if ws._worker_running:
-                    ui._enqueue(
-                        {
-                            "type": "busy_error",
-                            "message": "Cannot rewind/retry while processing.",
-                        }
+            verb = cmd_word[1:]
+            return JSONResponse(
+                {
+                    "error": (
+                        f"{cmd_word} is no longer served by /command; "
+                        f"use POST /v1/api/workstreams/{{ws_id}}/{verb}"
                     )
-                    return JSONResponse({"status": "busy"})
+                },
+                status_code=400,
+            )
 
         should_exit = ws.session.handle_command(cmd)
         if should_exit:
@@ -1318,65 +1408,6 @@ async def command(request: Request) -> JSONResponse:
         elif cmd_word == "/resume":
             # clear_ui signals the frontend to re-fetch history via REST.
             ui._enqueue({"type": "clear_ui"})
-        elif cmd_word in ("/rewind", "/retry"):
-            # clear_ui signals the frontend to re-fetch the (now
-            # truncated) history via REST and dispatch any queued
-            # edit-and-resend once it lands. Fires even on a rewind to
-            # zero messages: the frontend keys the resend off this
-            # signal, not an inline history payload.
-            ui._enqueue({"type": "clear_ui"})
-            # Audit trail
-            storage = getattr(request.app.state, "auth_storage", None)
-            if storage:
-                from turnstone.core.audit import record_audit
-
-                audit_uid, ip = _audit_context(request)
-                record_audit(
-                    storage,
-                    audit_uid,
-                    f"conversation.{cmd_word[1:]}",
-                    "workstream",
-                    ws.id,
-                    {"command": cmd, "ws_id": ws.id},
-                    ip,
-                )
-            # Dispatch deferred retry in background thread
-            retry_msg = ws.session._pending_retry
-            if retry_msg:
-                ws.session._pending_retry = None
-                session = ws.session
-
-                def run_retry() -> None:
-                    me = threading.current_thread()
-                    try:
-                        session.send(retry_msg)
-                    except GenerationCancelled:
-                        if ws.worker_thread is me:
-                            ui.on_stream_end()
-                            ui.on_state_change("idle")
-                    except Exception as exc:
-                        if ws.worker_thread is me:
-                            ui.on_error(f"Error: {exc}")
-                            ui.on_stream_end()
-                            ui.on_state_change("error")
-                    finally:
-                        with ws._lock:
-                            ws._worker_running = False
-
-                # Inlined rather than via ``session_worker.send`` because
-                # retry-when-busy is a hard reject (UI error, no fallback
-                # queue) — the shared dispatcher's enqueue/spawn shape
-                # doesn't fit. We gate on ``_worker_running`` for parity
-                # with that dispatcher so the two paths can't race into
-                # parallel workers on the same ChatSession.
-                with ws._lock:
-                    if ws._worker_running:
-                        ui.on_error("Cannot retry: workstream is busy")
-                    else:
-                        ws._worker_running = True
-                        t = threading.Thread(target=run_retry, daemon=True)
-                        ws.worker_thread = t
-                        t.start()
         # Sync in-memory workstream name after any command that can change it.
         # This ensures /api/workstreams and future page loads see the right name.
         if cmd_word in ("/name", "/resume"):
@@ -3706,6 +3737,17 @@ def create_app(
         accepted_permissions=("workstreams.close", "admin.coordinator"),
     )
     cancel_handler = make_cancel_handler(interactive_endpoint_config)
+    rewind_handler = make_rewind_handler(
+        interactive_endpoint_config,
+        audit_emit=_audit_rewind_workstream,
+        accepted_permissions=("conversation.modify",),
+    )
+    retry_handler = make_retry_handler(
+        interactive_endpoint_config,
+        dispatch_retry=_interactive_dispatch_retry,
+        audit_emit=_audit_retry_workstream,
+        accepted_permissions=("conversation.modify",),
+    )
     open_handler = make_open_handler(
         interactive_endpoint_config,
         audit_emit=_audit_workstream_opened,
@@ -3743,6 +3785,8 @@ def create_app(
             dequeue=dequeue_handler,  # lifted (P1.5) — DELETE /send
             approve=approve_handler,  # lifted: shared body
             cancel=cancel_handler,  # lifted: shared body
+            rewind=rewind_handler,  # lifted: shared body (#549)
+            retry=retry_handler,  # lifted: shared body (#549)
             events=events_handler,  # lifted: shared body
             history=history_handler,  # lifted: shared body (interactive feature gain)
             attachments=attachment_handlers,  # lifted: shared body (P1.5)

turnstone/ui/static/app.js

@@ -1314,33 +1314,44 @@ class Pane {
 
   _retryLast() {
     if (this.busy) return;
-    authFetch("/v1/api/command", {
-      method: "POST",
-      headers: { "Content-Type": "application/json" },
-      body: JSON.stringify({ command: "/retry", ws_id: this.wsId }),
-    }).catch((err) => {
+    // Path-keyed retry (#549). Truncation + re-dispatch happen
+    // server-side; the clear_ui event drives the history refetch.
+    authFetch(
+      "/v1/api/workstreams/" + encodeURIComponent(this.wsId) + "/retry",
+      { method: "POST" },
+    ).catch((err) => {
       this.addErrorMessage("Retry failed: " + err.message);
     });
   }
 
+  // Path-keyed rewind (#549) by absolute turn count. Shared by the
+  // per-message rewind button and the hand-typed /rewind reroute.
+  _rewindToTurns(turns) {
+    if (this.busy) return;
+    if (!Number.isInteger(turns) || turns < 1) return;
+    authFetch(
+      "/v1/api/workstreams/" + encodeURIComponent(this.wsId) + "/rewind",
+      {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({ turns }),
+      },
+    ).catch((err) => {
+      this.addErrorMessage("Rewind failed: " + err.message);
+    });
+  }
+
   _rewindToMessage(msgEl) {
     if (this.busy) return;
-    // Count how many user messages come at or after this one
+    // Count how many user messages come at or after this one. Bare
+    // ``.msg.user`` is intentional: system-nudge markers carry that
+    // class and the server's _find_turn_boundaries counts them as
+    // turns too, so this matches the server's rewind-N semantics.
     const userMsgs = this.messagesEl.querySelectorAll(".msg.user");
     const idx = Array.prototype.indexOf.call(userMsgs, msgEl);
     if (idx < 0) return;
     const turnsToRewind = userMsgs.length - idx;
-    if (turnsToRewind < 1) return;
-    authFetch("/v1/api/command", {
-      method: "POST",
-      headers: { "Content-Type": "application/json" },
-      body: JSON.stringify({
-        command: "/rewind " + turnsToRewind,
-        ws_id: this.wsId,
-      }),
-    }).catch((err) => {
-      this.addErrorMessage("Rewind failed: " + err.message);
-    });
+    this._rewindToTurns(turnsToRewind);
   }
 
   _startEdit(msgEl, originalText) {
@@ -1407,7 +1418,9 @@ class Pane {
 
   _editAndResend(msgEl, newText) {
     if (this.busy) return;
-    // Count turns to rewind (from this message onward)
+    // Count turns to rewind (from this message onward). Bare
+    // ``.msg.user`` matches the server's turn semantics — see
+    // _rewindToMessage.
     const userMsgs = this.messagesEl.querySelectorAll(".msg.user");
     const idx = Array.prototype.indexOf.call(userMsgs, msgEl);
     if (idx < 0) return;
@@ -1417,21 +1430,39 @@ class Pane {
     // Store pending send — dispatched from the clear_ui handler once
     // the rewind's truncated history is re-fetched over REST.
     this._pendingEditSend = newText;
-    authFetch("/v1/api/command", {
-      method: "POST",
-      headers: { "Content-Type": "application/json" },
-      body: JSON.stringify({
-        command: "/rewind " + turnsToRewind,
-        ws_id: this.wsId,
-      }),
-    })
-      .then((r) => {
+    authFetch(
+      "/v1/api/workstreams/" + encodeURIComponent(this.wsId) + "/rewind",
+      {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({ turns: turnsToRewind }),
+      },
+    )
+      .then(async (r) => {
         if (r && !r.ok) {
           this._pendingEditSend = null;
           this.setBusy(false);
           this.addErrorMessage(
             "Rewind failed (HTTP " + r.status + " " + r.statusText + ")",
           );
+          return;
+        }
+        // A 200 {"status":"busy"} means the rewind was rejected because a
+        // generation is in flight — no clear_ui fires, so the pending-edit
+        // latch + busy state would otherwise stay stuck (and the latch
+        // would later resend into the next clear_ui). Clear them here.
+        let data = null;
+        try {
+          data = await r.json();
+        } catch {
+          data = null;
+        }
+        if (data && data.status === "busy") {
+          this._pendingEditSend = null;
+          this.setBusy(false);
+          this.addErrorMessage(
+            "Cannot edit & resend while the workstream is processing.",
+          );
         }
       })
       .catch((err) => {
@@ -2050,6 +2081,28 @@ class Pane {
 
     if (text.startsWith("/")) {
       if (this.busy) return; // commands not allowed while busy
+      // /rewind and /retry were lifted to path-keyed endpoints (#549);
+      // reroute hand-typed ones so they don't 400 against /command.
+      const parts = text.split(/\s+/);
+      const cmdWord = parts[0].toLowerCase();
+      if (cmdWord === "/rewind") {
+        const n = parseInt(parts[1], 10);
+        if (!Number.isInteger(n) || n < 1) {
+          this.addErrorMessage(
+            "Usage: /rewind <N> — N must be a positive integer",
+          );
+          this.composer.clear();
+          return;
+        }
+        this._rewindToTurns(n);
+        this.composer.clear();
+        return;
+      }
+      if (cmdWord === "/retry") {
+        this._retryLast();
+        this.composer.clear();
+        return;
+      }
       authFetch("/v1/api/command", {
         method: "POST",
         headers: { "Content-Type": "application/json" },