Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions coworker/server/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -1707,6 +1707,10 @@ def _resolve_pending(resolution: str) -> None:
"command_trust": manager.workspace_command_trust(
str(getattr(engine, "audit_context", {}).get("workspace", ""))
),
# Mid-turn reconnect (#311): the client resets `running` on every
# session switch, and turn_start already fired — seed from the
# manager so Stop / Thinking / WaitingForAgent reappear.
"running": manager.is_running(session_id),
},
}
)
Expand Down
44 changes: 33 additions & 11 deletions surfaces/gui/e2e/fixtures.ts
Original file line number Diff line number Diff line change
Expand Up @@ -574,17 +574,38 @@ export async function mockApi(page: import("@playwright/test").Page) {
eventSockets.set(page, ws);
});

// Sessions mid-turn keep running after the viewing socket closes (real server
// behavior) so a reconnect can seed `ready.running` (#311).
const runningSessions = new Set<string>();
await page.routeWebSocket(/\/ws\/session\//, (ws) => {
const send = (type: string, data: Record<string, unknown> = {}) =>
ws.send(JSON.stringify({ type, data }));
send("ready");
const sid = (() => {
try {
const path = new URL(ws.url()).pathname;
return path.split("/").filter(Boolean).pop() || "";
} catch {
return "";
}
})();
const send = (type: string, data: Record<string, unknown> = {}) => {
try {
ws.send(JSON.stringify({ type, data }));
} catch {
/* socket may already be closed after a session switch */
}
};
const endTurn = () => {
runningSessions.delete(sid);
send("turn_done");
};
send("ready", { running: runningSessions.has(sid), session_id: sid });
let pendingTool = "run_shell"; // which proposal the next approval decision resolves
let epicTimer: ReturnType<typeof setInterval> | null = null; // the slow stream, stoppable via interrupt
let hadTurn = false; // a user_message landed — set_model is now a mid-session switch
ws.onMessage((raw) => {
const msg = JSON.parse(String(raw));
if (msg.type === "user_message") {
hadTurn = true;
runningSessions.add(sid);
send("turn_start", { input: msg.text });
if (/run a tool/i.test(msg.text)) {
pendingTool = "run_shell";
Expand Down Expand Up @@ -677,7 +698,7 @@ export async function mockApi(page: import("@playwright/test").Page) {
text: "Decision made.",
reasoning: thoughts.join(""),
});
send("turn_done");
endTurn();
}, 120);
return;
}
Expand All @@ -686,13 +707,13 @@ export async function mockApi(page: import("@playwright/test").Page) {
if (/compact the context/i.test(msg.text)) {
send("compacted", { text: "Context compacted — earlier turns were summarized" });
send("assistant_message", { text: "Still on it — continuing where I left off." });
send("turn_done");
endTurn();
return;
}
// A turn that dies on a provider error; the follow-up {type:"retry"} recovers.
if (/fail the turn/i.test(msg.text)) {
send("error", { error: "model unreachable" });
send("turn_done");
endTurn();
return;
}
// A deliberately SLOW multi-second stream (~40 ticks × 120ms) so specs can
Expand All @@ -708,7 +729,7 @@ export async function mockApi(page: import("@playwright/test").Page) {
clearInterval(epicTimer!);
epicTimer = null;
send("assistant_message", { text: ("The epic concludes. " + line).repeat(20) });
send("turn_done");
endTurn();
}
}, 120);
return;
Expand All @@ -729,7 +750,7 @@ export async function mockApi(page: import("@playwright/test").Page) {
cache_write: 800,
},
});
send("turn_done");
endTurn();
} else if (msg.type === "approval") {
if (pendingTool === "run_shell") {
if (msg.decision === "deny") {
Expand All @@ -747,7 +768,7 @@ export async function mockApi(page: import("@playwright/test").Page) {
// The decision echoes back so specs can pin what rode the wire (e.g. always_task).
send("assistant_message", { text: `Done via ${pendingTool} [decision=${msg.decision}]` });
}
send("turn_done");
endTurn();
} else if (msg.type === "interrupt") {
// Stop mid-stream: like the real engine, end the turn with `interrupted` and
// NO assistant_message — the client owns promoting the partial into the transcript.
Expand All @@ -756,7 +777,7 @@ export async function mockApi(page: import("@playwright/test").Page) {
epicTimer = null;
}
send("interrupted", {});
send("turn_done");
endTurn();
} else if (msg.type === "set_model") {
// Mid-session switch: the server applies it and broadcasts the persisted marker.
// Like the real server, the FIRST bind (fresh session) is silent.
Expand All @@ -767,9 +788,10 @@ export async function mockApi(page: import("@playwright/test").Page) {
});
} else if (msg.type === "retry") {
// Like the real engine: re-runs with NO new user message (turn_start input is empty).
runningSessions.add(sid);
send("turn_start", { input: "" });
send("assistant_message", { text: "Recovered after retry." });
send("turn_done");
endTurn();
}
});
});
Expand Down
24 changes: 24 additions & 0 deletions surfaces/gui/e2e/session-switch-running.spec.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,24 @@
// #311: switching away from a mid-turn session used to leave `running` stuck false on
// return (turn_start already fired). ready.running must restore Stop / live chrome.
import { expect } from "@playwright/test";
import { test } from "./fixtures";

test("switching back to a mid-turn session restores the Stop control", async ({ page }) => {
await page.goto("/");
await page.getByText("Draft the launch note").first().click();

// Park on an approval — the turn stays live (no turn_done) so ready.running stays true.
const box = page.getByPlaceholder(/Ask the coworker/);
await box.fill("run a tool");
await box.press("Enter");
await expect(page.getByRole("button", { name: "Stop" })).toBeVisible({ timeout: 10_000 });
await expect(page.getByText(/wants to run a command/i)).toBeVisible();

// Leave for another session while the turn is still live.
await page.getByText("Weekly plan 1").first().click();
await expect(page.getByRole("button", { name: "Stop" })).toHaveCount(0);

// Return — ready.running seeds the live chrome even though turn_start won't re-fire.
await page.getByText("Draft the launch note").first().click();
await expect(page.getByRole("button", { name: "Stop" })).toBeVisible({ timeout: 10_000 });
});
4 changes: 4 additions & 0 deletions surfaces/gui/src/App.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -584,6 +584,9 @@ export function App() {
if (d.command_trust?.required) setWorkspaceTrustRequest(d.command_trust);
// Cowork: adopt the server-provisioned scratch dir (only when we don't already have one).
if (d.workspace) setWorkspace((cur) => cur || d.workspace);
// Mid-turn reconnect (#311): selectSession clears `running`, and turn_start
// won't re-fire for an already-live turn — trust the server's claim.
if (typeof d.running === "boolean") setRunning(d.running);
break;
case "turn_start":
setRunning(true);
Expand Down Expand Up @@ -957,6 +960,7 @@ export function App() {
setSurface("session"); // selecting a conversation always returns to the conversation view
setTodo([]);
setStreaming("");
setReasoningStream("");
setRunning(false);
if (ag) setAgent(ag);
if (!gatesWorkspace(ag)) setShowGate(false);
Expand Down
17 changes: 17 additions & 0 deletions tests/test_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -323,6 +323,23 @@ def test_ws_simple_turn(tmp_path):
assert "turn_end" in types


def test_ws_ready_reports_running_for_mid_turn_reconnect(tmp_path):
# #311: a client that reconnects mid-turn must learn the session is still live —
# turn_start already fired and won't re-fire for that turn.
manager = SessionManager(workspace=tmp_path, provider=ScriptedProvider([]))
manager.mark_running("mid-turn")
client = TestClient(create_app(manager))
with client.websocket_connect("/ws/session/mid-turn") as ws:
ready = ws.receive_json()
assert ready["type"] == "ready"
assert ready["data"]["running"] is True
manager.mark_idle("mid-turn")
with client.websocket_connect("/ws/session/mid-turn") as ws:
ready = ws.receive_json()
assert ready["type"] == "ready"
assert ready["data"]["running"] is False


def test_ws_rejects_oversized_message(tmp_path):
from coworker.server import app as app_mod
from coworker.attachments import MAX_ATTACHMENTS
Expand Down