diff --git a/sdk/typescript/src/server.ts b/sdk/typescript/src/server.ts index b981106c..1f6ef711 100644 --- a/sdk/typescript/src/server.ts +++ b/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 { + 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 { + return this.request( + "POST", + `/v1/api/workstreams/${encodeURIComponent(wsId)}/retry`, + { json: {} }, + ); + } + // -- Streaming ------------------------------------------------------------ async *streamEvents(wsId: string): AsyncIterableIterator { diff --git a/tests/test_auth.py b/tests/test_auth.py index f1fcb0a8..c4316498 100644 --- a/tests/test_auth.py +++ b/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" diff --git a/tests/test_coordinator_client.py b/tests/test_coordinator_client.py index 0ad41627..a0d0a9fb 100644 --- a/tests/test_coordinator_client.py +++ b/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 diff --git a/tests/test_openapi.py b/tests/test_openapi.py index 2c86c67b..03ff500e 100644 --- a/tests/test_openapi.py +++ b/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", diff --git a/tests/test_route_proxy_audit.py b/tests/test_route_proxy_audit.py index d7ca45dd..0bea4f34 100644 --- a/tests/test_route_proxy_audit.py +++ b/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"), diff --git a/tests/test_sdk_console.py b/tests/test_sdk_console.py index b6e5c6ac..2e6afe3b 100644 --- a/tests/test_sdk_console.py +++ b/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 # --------------------------------------------------------------------------- diff --git a/tests/test_sdk_server.py b/tests/test_sdk_server.py index 6e19b407..698186ac 100644 --- a/tests/test_sdk_server.py +++ b/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 # --------------------------------------------------------------------------- diff --git a/tests/test_session_routes.py b/tests/test_session_routes.py index f03d1d4c..21fea947 100644 --- a/tests/test_session_routes.py +++ b/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 diff --git a/tests/test_workstream_endpoints.py b/tests/test_workstream_endpoints.py index 75c6f7be..dee3a701 100644 --- a/tests/test_workstream_endpoints.py +++ b/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 # =========================================================================== diff --git a/turnstone/api/console_spec.py b/turnstone/api/console_spec.py index 582e82e6..508eca54 100644 --- a/turnstone/api/console_spec.py +++ b/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", diff --git a/turnstone/api/server_schemas.py b/turnstone/api/server_schemas.py index d979e7d0..bb3bbd0c 100644 --- a/turnstone/api/server_schemas.py +++ b/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") diff --git a/turnstone/api/server_spec.py b/turnstone/api/server_spec.py index 6dd3906c..004ad9b2 100644 --- a/turnstone/api/server_spec.py +++ b/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, diff --git a/turnstone/console/coordinator_client.py b/turnstone/console/coordinator_client.py index d5df881a..74a0f121 100644 --- a/turnstone/console/coordinator_client.py +++ b/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. diff --git a/turnstone/console/server.py b/turnstone/console/server.py index ebdb110a..e0323910 100644 --- a/turnstone/console/server.py +++ b/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( diff --git a/turnstone/core/auth.py b/turnstone/core/auth.py index 2dcf59a3..ec19d6e1 100644 --- a/turnstone/core/auth.py +++ b/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" diff --git a/turnstone/core/session_routes.py b/turnstone/core/session_routes.py index d59856d6..ab7508a4 100644 --- a/turnstone/core/session_routes.py +++ b/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, *, diff --git a/turnstone/sdk/console.py b/turnstone/sdk/console.py index c3fb4b46..07bcf0f7 100644 --- a/turnstone/sdk/console.py +++ b/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)) diff --git a/turnstone/sdk/server.py b/turnstone/sdk/server.py index f411279f..17056677 100644 --- a/turnstone/sdk/server.py +++ b/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]: diff --git a/turnstone/server.py b/turnstone/server.py index b69aadcf..5a92198f 100644 --- a/turnstone/server.py +++ b/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) diff --git a/turnstone/ui/static/app.js b/turnstone/ui/static/app.js index aebf6a00..f36b1f91 100644 --- a/turnstone/ui/static/app.js +++ b/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 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" },