From aafb4f429183bc0daa5f632fb0496799d21591f9 Mon Sep 17 00:00:00 2001 From: Rohit C Prasad Date: Mon, 17 Aug 2026 16:05:16 -0700 Subject: [PATCH] Feed model: interest follows the assignment relation; notes are pure appends MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Worker delivery is now a read-time feed over its slice (assigned ∪ filed) — send-backs, comment answers, reassignments, cancels, and acceptance all arrive through one relation; per-event recipient addressing retired. Reassignment delivers before interest ends; a new assignee replays the item's story. Detail pane gains Add a note (never changes state); external pending/consume become space-scoped feed calls. --- coworker/server/app.py | 19 ++++- coworker/server/manager.py | 63 ++++++++++++++-- coworker/teams/cli.py | 4 +- coworker/teams/dialect.py | 22 +++--- coworker/teams/mcp_server.py | 13 ++-- coworker/teams/store.py | 84 ++++++++++++---------- surfaces/gui/e2e/board.spec.ts | 17 +++++ surfaces/gui/e2e/fixtures.ts | 17 ++++- surfaces/gui/src/App.tsx | 2 + surfaces/gui/src/api.ts | 17 +++++ surfaces/gui/src/components/BoardPanel.tsx | 44 ++++++++++++ surfaces/gui/src/styles.css | 7 ++ tests/test_team_open_surface.py | 10 ++- tests/test_team_store.py | 11 ++- tests/test_team_wake.py | 73 +++++++++++++------ 15 files changed, 305 insertions(+), 98 deletions(-) diff --git a/coworker/server/app.py b/coworker/server/app.py index 1123b2aa..ffceba1a 100644 --- a/coworker/server/app.py +++ b/coworker/server/app.py @@ -763,6 +763,13 @@ def create_app(manager: SessionManager) -> FastAPI: media_type=manager.attachment_store.mime_for(name), ) + @app.post("/v1/sessions/{session_id}/board/comment") + def session_board_comment(session_id: str, body: dict) -> dict[str, Any]: + body = body or {} + return manager.board_comment( + session_id, int(body.get("item", 0)), str(body.get("body", "")) + ) + @app.post("/v1/sessions/{session_id}/board/transition") def session_board_transition(session_id: str, body: dict) -> dict[str, Any]: body = body or {} @@ -999,11 +1006,15 @@ def create_app(manager: SessionManager) -> FastAPI: ) @app.get("/v1/board/pending") - def board_pending(request: Request, limit: int = 200): + def board_pending(request: Request, space: str, limit: int = 200): + # The actor's FEED: events on its slice since its cursor — interest + # follows the assignment relation, same projection in-app workers use. return _board( request, lambda actor: { - "events": manager.team_store.pending_for(actor.id, limit=int(limit)) + "events": manager.team_store.feed_for( + space, actor.id, limit=int(limit) + ) }, ) @@ -1012,7 +1023,9 @@ def create_app(manager: SessionManager) -> FastAPI: body = body or {} def run(actor): - manager.team_store.consume(actor.id, int(body.get("upto_seq", 0))) + manager.team_store.consume_feed( + str(body.get("space", "")), actor.id, int(body.get("upto_seq", 0)) + ) return {"ok": True} return _board(request, run) diff --git a/coworker/server/manager.py b/coworker/server/manager.py index 44c738b6..f63a9ee3 100644 --- a/coworker/server/manager.py +++ b/coworker/server/manager.py @@ -1458,6 +1458,21 @@ class SessionManager: item["timeline"] = timeline return item + def board_comment(self, session_id: str, item_id: int, body: str) -> dict[str, Any]: + """A pure note from the user on an item — never changes state (owner + doctrine 2026-08-17); the assignee hears it through its feed.""" + space = self._board_space(session_id) + if space is None: + return {"error": "no board for this session"} + try: + event = self.team_store.comment( + space, self._user_actor(), int(item_id), body + ) + except (TeamsBoardError, ValueError) as error: + return {"error": str(error)} + self.kick_team_tick() # the assignee's feed has news + return {"ok": True, "seq": event["seq"]} + def session_board(self, session_id: str) -> dict[str, Any]: """The session's board: items grouped by the workspace-keyed space. Empty (space=None) when the workspace has no items — the rail hides itself.""" @@ -1983,10 +1998,16 @@ class SessionManager: async def _drain_team_member( self, team, *, session_id: str, actor: str, is_lead: bool ) -> int: - directs = self.team_store.pending_for(actor) + # Interest follows the assignment relation: everyone's feed is the events + # on their slice (assigned ∪ filed) — comments, moves, reassignments. The + # lead additionally subscribes to the board-wide decision classes. + directs = self.team_store.feed_for(team.space, actor) subs = ( self.team_store.subscribed_events(team.space, actor) if is_lead else [] ) + if subs: + seen = {e["seq"] for e in subs} + directs = [e for e in directs if e["seq"] not in seen] chat_handle = "lead" if is_lead else actor chats = ( self.chat_store.unread_for(team.chat_group, chat_handle) @@ -1994,12 +2015,21 @@ class SessionManager: else [] ) # Cancel is top-priority: an in-flight worker gets interrupted NOW; the - # queued notice (delivered when the turn dies) tells it why. + # queued notice (delivered when the turn dies) tells it why. Only for the + # item's ASSIGNEE — a filer merely hears about it. + def _holds(event) -> bool: + try: + item = self.team_store.get_item(team.space, int(event["item_id"])) + except Exception: + return False + return item["assignee"] == actor + cancels = [ e for e in directs if e["kind"] == "item_transitioned" and (e.get("payload") or {}).get("to") == "canceled" + and _holds(e) ] if cancels and self.is_running(session_id): engine = self._engines.get(session_id) @@ -2012,7 +2042,9 @@ class SessionManager: if not self.teams.count_wake(team.team_id, cap=self.TEAM_WAKE_CAP_PER_HOUR): logger.warning("team %s paused for budget this hour", team.team_id) return 0 - message, rows = self._team_digest(team, directs, subs, chats, is_lead=is_lead) + message, rows = self._team_digest( + team, directs, subs, chats, is_lead=is_lead, reader=actor + ) self._team_inflight.add(session_id) source = self._board_source(team, message, rows=rows) @@ -2021,8 +2053,11 @@ class SessionManager: await self.deliver_to_session(session_id, message, source=source) # Consume only after the turn dispatched: a crash before this replays # the batch next tick (at-least-once, never silently lost). - if directs: - self.team_store.consume(actor, directs[-1]["seq"]) + # The feed cursor advances past BOTH batches: a subs event deduped + # out of directs must not replay as a direct next tick. + delivered = [e["seq"] for e in directs] + [e["seq"] for e in subs] + if delivered: + self.team_store.consume_feed(team.space, actor, max(delivered)) if subs: self.team_store.consume_subscription( team.space, actor, subs[-1]["seq"] @@ -2060,6 +2095,7 @@ class SessionManager: chats: Optional[list[dict]] = None, *, is_lead: bool, + reader: str = "", ) -> tuple[str, list[dict]]: """Coalesce one queue batch into one wake message. Deterministic, computed by code — the model does judgment, not arithmetic. Returns (model text, @@ -2088,6 +2124,7 @@ class SessionManager: if event["kind"] == "item_assigned": if item is None: continue + assignee = payload.get("assignee") or "" if payload.get("claimed"): # A self-claim surfacing in the lead's subscription feed — # supervision by exception, not an assignment to the reader. @@ -2097,12 +2134,26 @@ class SessionManager: ) rows.append({**row, "kind": "claimed"}) continue + if reader and payload.get("previous") == reader and assignee != reader: + # The reader just LOST this item — its interest ends here. + lines.append( + f"{title} was reassigned to {assignee} by {event['actor']}" + " — stop any work on it; hand off context via a comment" + " if useful." + ) + rows.append({**row, "kind": "assigned", "assignee": assignee}) + continue + if reader and assignee != reader: + # Someone else's assignment surfacing in a broader feed. + lines.append(f"{title} assigned to {assignee} by {event['actor']}") + rows.append({**row, "kind": "assigned", "assignee": assignee}) + continue lines.append( f"You've been assigned work item {title}.\n" f" Done when: {item['criteria']}" + (f"\n Details: {item['description']}" if item["description"] else "") ) - rows.append({**row, "kind": "assigned"}) + rows.append({**row, "kind": "assigned", "assignee": assignee}) elif event["kind"] == "item_transitioned": to = payload.get("to", "?") comment = clamp(payload.get("comment") or "") diff --git a/coworker/teams/cli.py b/coworker/teams/cli.py index 666030dc..6bb5de26 100644 --- a/coworker/teams/cli.py +++ b/coworker/teams/cli.py @@ -384,7 +384,7 @@ def _cmd_policy(args) -> int: def _cmd_pending(args) -> int: dialect = _dialect(args) - events = dialect.pending(limit=args.limit) + events = dialect.pending(_space(args), limit=args.limit) if args.json: print(json.dumps(events, indent=2)) else: @@ -394,7 +394,7 @@ def _cmd_pending(args) -> int: if not events: print("nothing pending") if args.consume and events: - dialect.consume(events[-1]["seq"]) + dialect.consume(_space(args), events[-1]["seq"]) return 0 diff --git a/coworker/teams/dialect.py b/coworker/teams/dialect.py index 540b0806..1bdb4d78 100644 --- a/coworker/teams/dialect.py +++ b/coworker/teams/dialect.py @@ -90,8 +90,8 @@ class BoardDialect(Protocol): def attachment(self, stored: str) -> tuple[bytes, str]: ... def policy(self, space: str) -> dict[str, Any]: ... def set_policy(self, space: str, *, claims: str) -> dict[str, Any]: ... - def pending(self, *, limit: int = 200) -> list[dict[str, Any]]: ... - def consume(self, upto_seq: int) -> None: ... + def pending(self, space: str, *, limit: int = 200) -> list[dict[str, Any]]: ... + def consume(self, space: str, upto_seq: int) -> None: ... def journal_append( self, case: str, @@ -237,11 +237,11 @@ class LocalDialect: def set_policy(self, space: str, *, claims: str) -> dict[str, Any]: return self.store.set_policy(space, self.actor, claims=claims) - def pending(self, *, limit: int = 200) -> list[dict[str, Any]]: - return self.store.pending_for(self.actor.id, limit=limit) + def pending(self, space: str, *, limit: int = 200) -> list[dict[str, Any]]: + return self.store.feed_for(space, self.actor.id, limit=limit) - def consume(self, upto_seq: int) -> None: - self.store.consume(self.actor.id, int(upto_seq)) + def consume(self, space: str, upto_seq: int) -> None: + self.store.consume_feed(space, self.actor.id, int(upto_seq)) def journal_append( self, @@ -472,11 +472,13 @@ class RemoteDialect: def set_policy(self, space: str, *, claims: str) -> dict[str, Any]: return self._post("/v1/board/policy", {"space": space, "claims": claims}) - def pending(self, *, limit: int = 200) -> list[dict[str, Any]]: - return self._get("/v1/board/pending", {"limit": limit})["events"] + def pending(self, space: str, *, limit: int = 200) -> list[dict[str, Any]]: + return self._get("/v1/board/pending", {"space": space, "limit": limit})[ + "events" + ] - def consume(self, upto_seq: int) -> None: - self._post("/v1/board/consume", {"upto_seq": int(upto_seq)}) + def consume(self, space: str, upto_seq: int) -> None: + self._post("/v1/board/consume", {"space": space, "upto_seq": int(upto_seq)}) def journal_append( self, diff --git a/coworker/teams/mcp_server.py b/coworker/teams/mcp_server.py index d3f67c31..1f9466a8 100644 --- a/coworker/teams/mcp_server.py +++ b/coworker/teams/mcp_server.py @@ -120,16 +120,17 @@ def build(dialect, *, space: str): @mcp.tool() def board_pending() -> Any: - """Your unconsumed deliveries — assignments addressed to you, cancel - notices. Check at the start of a work session; acknowledge with - board_consume.""" - return _safe(dialect.pending) + """Your unconsumed feed: every event on items assigned to you or filed + by you — assignments, send-backs with feedback, comments from the lead + or user, cancellations. Check at the start of a work session and before + finishing; acknowledge with board_consume.""" + return _safe(dialect.pending, space) @mcp.tool() def board_consume(upto_seq: int) -> Any: - """Acknowledge deliveries up to a sequence number (from board_pending), + """Acknowledge feed events up to a sequence number (from board_pending), so they are not re-delivered.""" - return _safe(lambda: (dialect.consume(upto_seq), {"ok": True})[1]) + return _safe(lambda: (dialect.consume(space, upto_seq), {"ok": True})[1]) if role in ("lead", "user"): diff --git a/coworker/teams/store.py b/coworker/teams/store.py index 0362688a..df52e529 100644 --- a/coworker/teams/store.py +++ b/coworker/teams/store.py @@ -15,8 +15,10 @@ Mechanics, kept boring: detects out-of-band edits. Tamper-evidence, not tamper-proofing. - `taint` marks records authored after touching untrusted content; readers render it as provenance ("treat as evidence, not instructions"). -- `recipient` is how per-agent delivery works: a projection over the one log, not a - second write path (consumption semantics arrive with the wake plumbing). +- Per-agent delivery is the FEED projection over the one log (never a second write + path): interest follows the assignment relation — a worker is subscribed to its + slice, cursors mark consumption. The `recipient` column is retired plumbing + (kept in the schema; no longer written). """ from __future__ import annotations @@ -290,22 +292,44 @@ class TeamStore: ).fetchall() return [_row_to_event(row) for row in rows] - # -------------------------------------------------- delivery (durable queue) + # -------------------------------------------------- delivery (durable feed) - # The per-agent durable queue is a PROJECTION over the one log, never a second - # write path: entries are events addressed to a recipient, "consumed" is a - # cursor. Durable-until-consumed (a crash before consume() replays on the next - # drain); coalescing happens at dequeue — the caller turns one batch into one - # digest. "Mailbox" is banned as a concept; this is internal plumbing. + # The per-agent durable feed is a PROJECTION over the one log, never a second + # write path — and INTEREST FOLLOWS THE ASSIGNMENT RELATION (owner ruling + # 2026-08-17): a worker is subscribed to events on its slice (items assigned + # to it or filed by it — subscription ≡ visibility, one boundary), with no + # per-event addressing decisions in the write path. "Consumed" is a cursor; + # durable-until-consumed (a crash before consume replays on the next drain); + # coalescing happens at dequeue. "Mailbox" is banned as a concept. - def pending_for(self, recipient: str, *, limit: int = 200) -> list[dict[str, Any]]: - """Unconsumed events addressed to this agent, in order.""" - return self.for_recipient( - recipient, since_seq=self._cursor(f"to:{recipient}"), limit=limit - ) + def feed_for( + self, space: str, actor_id: str, *, limit: int = 200 + ) -> list[dict[str, Any]]: + """Unconsumed events this actor is subscribed to, in order: everything on + its current slice, plus assignment events that START its interest (newly + assigned to it) or END it (just reassigned away — it hears that, then + goes quiet). Its own events never appear.""" + key = f"feed:{actor_id}:{space}" + events = self.events(space, since_seq=self._cursor(key), limit=limit) + with self._lock: + slice_ids = self._worker_slice(space, actor_id) + out = [] + for event in events: + if event["actor"] == actor_id: + continue + payload = event.get("payload") or {} + if event["kind"] == ITEM_ASSIGNED and actor_id in ( + payload.get("assignee"), + payload.get("previous"), + ): + out.append(event) + continue + if event.get("item_id") in slice_ids: + out.append(event) + return out - def consume(self, recipient: str, upto_seq: int) -> None: - self._set_cursor(f"to:{recipient}", upto_seq) + def consume_feed(self, space: str, actor_id: str, upto_seq: int) -> None: + self._set_cursor(f"feed:{actor_id}:{space}", int(upto_seq)) # Lead subscriptions: an ALLOWLIST of decision-demanding event classes — a # worker moving its item to review/blocked, or filing a new item. Journal @@ -545,32 +569,16 @@ class TeamStore: f"illegal transition {current.value} → {target.value}" ) self._check_transition_authority(actor, item, current, target) - # When someone ELSE moves your assigned item to a state that needs - # YOUR action, the event is addressed to you: a send-back - # (review→in_progress with feedback), an unblock, a cancel (whose - # delivery/in-flight interrupt makes the worker actually stop). This - # is a board write, not a message — delivery is the queue projection - # doing its job. DONE is deliberately unaddressed: waking a worker to - # say its finished item is finished would burn a turn for nothing. - action_needed = { - ItemState.IN_PROGRESS, - ItemState.BLOCKED, - ItemState.CANCELED, - } - recipient = ( - item["assignee"] - if target in action_needed - and item["assignee"] - and item["assignee"] != actor.id - else None - ) + # No per-event addressing: delivery is the FEED projection — interest + # follows the assignment relation (see feed_for), so a send-back, an + # unblock, a cancel, or an acceptance reaches whoever holds the item + # without the store editorializing about who cares. event = self.append_event( space, ITEM_TRANSITIONED, actor, item_id=item_id, case_id=item["case_id"] or None, - recipient=recipient, payload={ "from": current.value, "to": target.value, @@ -615,8 +623,9 @@ class TeamStore: def assign( self, space: str, actor: Actor, item_id: int, assignee: str ) -> dict[str, Any]: - """Set the assignee. Not a message: the event addresses the assignee - (`recipient`), and the worker's prompt derives from the item itself.""" + """Set the assignee. Not a message: the feed projection delivers it — the + new assignee's interest starts with this event, and the previous + assignee's interest ends with it (both hear it; see feed_for).""" self._require(actor, {Role.USER, Role.LEAD}, "assign") if not (assignee or "").strip(): raise BoardError("assignee is required") @@ -633,7 +642,6 @@ class TeamStore: actor, item_id=item_id, case_id=item["case_id"] or None, - recipient=assignee, payload={"assignee": assignee, "previous": item["assignee"] or ""}, ) if self.journal is not None and item["case_id"]: diff --git a/surfaces/gui/e2e/board.spec.ts b/surfaces/gui/e2e/board.spec.ts index b7252d60..72d7d2d8 100644 --- a/surfaces/gui/e2e/board.spec.ts +++ b/surfaces/gui/e2e/board.spec.ts @@ -118,6 +118,23 @@ test("item detail: timeline with attachment, worker link, request changes", asyn await expect(detail).toContainText("Dependency audit — lockfiles"); }); +test("Add a note is a pure append — it lands in the timeline, state untouched", async ({ + page, +}) => { + await planTheWork(page); + await page.getByTestId("board-rail").getByText("Report rollup").click(); + const detail = page.getByTestId("board-detail"); + await expect(detail).toContainText("In review"); + await detail.getByTestId("board-note-input").fill("prefer the v2 endpoint for totals"); + await detail.getByTestId("board-note-input").press("Enter"); + // the note appears as a timeline event… + await expect(detail).toContainText("user commented"); + await expect(detail).toContainText("prefer the v2 endpoint for totals"); + // …and the state did NOT change (notes never transition) + await expect(detail).toContainText("In review"); + await expect(detail.getByRole("button", { name: "Mark done" })).toBeVisible(); +}); + test("journal section lists cases once a board exists", async ({ page }) => { await planTheWork(page); await page.getByRole("button", { name: /Journal/ }).click(); diff --git a/surfaces/gui/e2e/fixtures.ts b/surfaces/gui/e2e/fixtures.ts index 774e72f3..a452a8cc 100644 --- a/surfaces/gui/e2e/fixtures.ts +++ b/surfaces/gui/e2e/fixtures.ts @@ -576,6 +576,9 @@ export async function mockApi(page: import("@playwright/test").Page) { mentions: ["nia"], }, ]; + // Pure notes added from the detail pane (never change state) — appended to + // the item's timeline so the pane reflects them after reload. + const itemNotes: Record = {}; const seedBoard = () => { if (boardItems.length) return; boardItems.push( @@ -1099,7 +1102,19 @@ export async function mockApi(page: import("@playwright/test").Page) { }, ] : [{ seq: 30, ts: at, actor: "lead", kind: "created" }]; - return json({ ...item, timeline }); + return json({ ...item, timeline: timeline.concat(itemNotes[id] || []) }); + } + if (/\/v1\/sessions\/[^/]+\/board\/comment$/.test(p) && m === "POST") { + const b = req.postDataJSON() || {}; + const id = Number(b.item); + (itemNotes[id] = itemNotes[id] || []).push({ + seq: 90 + (itemNotes[id]?.length || 0), + ts: new Date().toISOString(), + actor: "user", + kind: "comment", + body: String(b.body || ""), + }); + return json({ ok: true }); } if (/\/v1\/sessions\/[^/]+\/board\/attachment$/.test(p)) { // A real 1x1 PNG so the actually loads (the spec asserts it renders). diff --git a/surfaces/gui/src/App.tsx b/surfaces/gui/src/App.tsx index 54f9533b..5b1f81f4 100644 --- a/surfaces/gui/src/App.tsx +++ b/surfaces/gui/src/App.tsx @@ -3,6 +3,7 @@ import { announceInboxUnlock, createTempWorkspace, finalizeAutomationRun, + boardComment, boardTransition, fetchBoardAttachment, getBoardItem, @@ -2075,6 +2076,7 @@ export function App() { setBoardDetailId(null); }} onTransition={moveBoardItem} + onComment={(item, body) => boardComment(sessionId, item, body)} loadItem={(id) => getBoardItem(sessionId, id)} loadAttachment={(stored) => fetchBoardAttachment(sessionId, stored)} onOpenWorker={(actor) => { diff --git a/surfaces/gui/src/api.ts b/surfaces/gui/src/api.ts index 519ce5fe..f1b0dce1 100644 --- a/surfaces/gui/src/api.ts +++ b/surfaces/gui/src/api.ts @@ -293,6 +293,23 @@ export async function fetchBoardAttachment( return URL.createObjectURL(await res.blob()); } +// A pure note on an item — never changes state; the assignee hears it via its feed. +export async function boardComment( + sessionId: string, + item: number, + body: string, +): Promise<{ ok?: boolean; error?: string }> { + const res = await fetch( + `${httpBase()}/v1/sessions/${encodeURIComponent(sessionId)}/board/comment`, + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ item, body }), + }, + ); + return res.json(); +} + export async function boardTransition( sessionId: string, item: number, diff --git a/surfaces/gui/src/components/BoardPanel.tsx b/surfaces/gui/src/components/BoardPanel.tsx index c6ab5937..95e7fdc1 100644 --- a/surfaces/gui/src/components/BoardPanel.tsx +++ b/surfaces/gui/src/components/BoardPanel.tsx @@ -122,6 +122,7 @@ export function BoardOverlay({ board, onClose, onTransition, + onComment, loadItem, loadAttachment, onOpenWorker, @@ -131,6 +132,8 @@ export function BoardOverlay({ onClose: () => void; // (item, to, comment?) → performed as the user; App refetches on completion. onTransition?: (item: number, to: string, comment?: string) => void; + // A pure note — never changes state; the assignee hears it through its feed. + onComment?: (item: number, body: string) => Promise | void; loadItem?: (id: number) => Promise; loadAttachment?: (stored: string) => Promise; // Assignee link → jump into that coworker's session (closes the overlay). @@ -162,6 +165,10 @@ export function BoardOverlay({ // the pane refreshes on the next tick so the transition's board refetch lands first if (detail?.id === item) setTimeout(() => void openItem(item), 350); }; + const addNote = async (item: number, body: string) => { + await onComment?.(item, body); + await openItem(item); + }; const finished = board.items.filter( (i) => i.state === "done" || i.state === "canceled" @@ -238,6 +245,7 @@ export function BoardOverlay({ @@ -260,11 +268,13 @@ const STATE_LABEL: Record = { function ItemDetail({ detail, onTransition, + onAddNote, loadAttachment, onOpenWorker, }: { detail: BoardItemDetail; onTransition?: (item: number, to: string, comment?: string) => void; + onAddNote?: (item: number, body: string) => Promise; loadAttachment?: (stored: string) => Promise; onOpenWorker?: (actor: string) => void; }) { @@ -317,6 +327,7 @@ function ItemDetail({ ))} + {onAddNote && } {onTransition && ( Promise; +}) { + const [text, setText] = useState(""); + useEffect(() => setText(""), [detail.id]); + const submit = async () => { + const body = text.trim(); + if (!body) return; + setText(""); + await onAddNote(detail.id, body); + }; + return ( + setText(e.target.value)} + onKeyDown={(e) => { + if (e.key === "Enter") void submit(); + }} + /> + ); +} + function DetailActions({ detail, onTransition, diff --git a/surfaces/gui/src/styles.css b/surfaces/gui/src/styles.css index 4da6b81c..8f8d2101 100644 --- a/surfaces/gui/src/styles.css +++ b/surfaces/gui/src/styles.css @@ -1784,6 +1784,13 @@ html[data-platform="linux"] ::-webkit-scrollbar-thumb:hover { background-color: .board-shot img { display: block; width: 100%; max-height: 120px; object-fit: cover; background: var(--paper); } .board-shot-cap { display: block; font-size: 10.5px; color: var(--faint); padding: 4px 8px; background: var(--panel); } +.board-note-input { + width: 100%; margin-top: 14px; font: inherit; font-size: 12.5px; color: var(--ink); + background: var(--paper); border: 1px solid var(--line); border-radius: 999px; + padding: 7px 14px; outline: none; +} +.board-note-input::placeholder { color: var(--faint); } +.board-note-input:focus { border-color: var(--accent); background: var(--panel); } .board-detail-actions { display: flex; gap: 14px; align-items: center; margin-top: 20px; } .board-btn { font-size: 12.5px; cursor: pointer; font-family: inherit; } .board-btn.primary { diff --git a/tests/test_team_open_surface.py b/tests/test_team_open_surface.py index 9f90b860..ad7b5d40 100644 --- a/tests/test_team_open_surface.py +++ b/tests/test_team_open_surface.py @@ -313,10 +313,14 @@ def test_pending_and_consume_over_the_wire(api): ) item = lead.create_item("proj", title="Queued", criteria="c") lead.assign("proj", item["id"], "nia") - events = nia.pending() + events = nia.pending("proj") assert events and events[-1]["kind"] == "item_assigned" - nia.consume(events[-1]["seq"]) - assert nia.pending() == [] + nia.consume("proj", events[-1]["seq"]) + assert nia.pending("proj") == [] + # a comment ANSWER arrives through the same feed — no addressing anywhere + lead.comment("proj", item["id"], "start with the v2 endpoint") + follow = nia.pending("proj") + assert [e["kind"] for e in follow] == ["item_commented"] # ------------------------------------------------------------------ attachments diff --git a/tests/test_team_store.py b/tests/test_team_store.py index 4d6494c9..2269ca88 100644 --- a/tests/test_team_store.py +++ b/tests/test_team_store.py @@ -73,14 +73,13 @@ def test_spaces_created_lazily_and_isolated(store): assert [i["id"] for i in store.list_items("beta", USER)] == [1] -def test_assignment_lands_in_the_assignees_deliveries(store): +def test_assignment_lands_in_the_assignees_feed(store): item = seed(store) store.assign("proj", LEAD, item["id"], "worker-1") - deliveries = store.for_recipient("worker-1") - assert len(deliveries) == 1 - assert deliveries[0]["kind"] == "item_assigned" - assert deliveries[0]["payload"]["assignee"] == "worker-1" - assert store.for_recipient("worker-2") == [] + deliveries = store.feed_for("proj", "worker-1") + assert [e["kind"] for e in deliveries] == ["item_created", "item_assigned"] + assert deliveries[-1]["payload"]["assignee"] == "worker-1" + assert store.feed_for("proj", "worker-2") == [] def test_rebuild_reproduces_the_projection(store): diff --git a/tests/test_team_wake.py b/tests/test_team_wake.py index c7057c8b..7205c0cb 100644 --- a/tests/test_team_wake.py +++ b/tests/test_team_wake.py @@ -61,17 +61,18 @@ def assigned(store, assignee="swe-worker"): return item["id"] -def test_deliveries_are_durable_until_consumed(store): +def test_feed_is_durable_until_consumed(store): assigned(store) - first = store.pending_for("swe-worker") - assert len(first) == 1 and first[0]["kind"] == "item_assigned" + first = store.feed_for(SPACE, "swe-worker") + # the item's creation and the assignment both start the worker's story + assert [e["kind"] for e in first] == ["item_created", "item_assigned"] # not consumed → still pending (crash-safe replay) - assert store.pending_for("swe-worker") == first - store.consume("swe-worker", first[-1]["seq"]) - assert store.pending_for("swe-worker") == [] + assert store.feed_for(SPACE, "swe-worker") == first + store.consume_feed(SPACE, "swe-worker", first[-1]["seq"]) + assert store.feed_for(SPACE, "swe-worker") == [] # a second assignment queues fresh assigned(store) - assert len(store.pending_for("swe-worker")) == 1 + assert len(store.feed_for(SPACE, "swe-worker")) == 2 def test_lead_subscriptions_are_an_allowlist(store): @@ -408,32 +409,58 @@ def test_item_detail_timeline_and_blocker_fact(manager): assert blocked["blocker"] == "need the staging tfvars" -def test_transitions_by_others_address_the_assignee(store): - """A send-back or unblock is a board write, not a message — but the queue - projection addresses it to the worker whose action it demands (owner - question 2026-08-17 exposed the gap: only cancel was addressed).""" +def test_feed_interest_follows_the_assignment_relation(store): + """Owner ruling 2026-08-17: no per-event addressing — a worker is subscribed + to everything on its slice. Send-backs, comment ANSWERS (the silently broken + path), and acceptance all arrive through one relation.""" item_id = assigned(store) - store.consume("swe-worker", store.pending_for("swe-worker")[-1]["seq"]) + store.consume_feed(SPACE, "swe-worker", store.feed_for(SPACE, "swe-worker")[-1]["seq"]) store.transition(SPACE, WORKER, item_id, "in_progress") store.transition(SPACE, WORKER, item_id, "review", comment="ready") - assert store.pending_for("swe-worker") == [] # own moves never self-address - # the lead's send-back reaches the worker's queue, feedback attached + assert store.feed_for(SPACE, "swe-worker") == [] # own moves never self-deliver + # the lead's send-back arrives with the feedback… store.transition(SPACE, LEAD, item_id, "in_progress", comment="totals drift") - pending = store.pending_for("swe-worker") - assert [e["payload"]["to"] for e in pending] == ["in_progress"] - assert pending[-1]["payload"]["comment"] == "totals drift" - store.consume("swe-worker", pending[-1]["seq"]) - # ...but done is deliberately unaddressed — no wake to hear "it's finished" + # …and so does a lead's comment ANSWER (never delivered under addressing) + store.comment(SPACE, LEAD, item_id, "use the v2 endpoint") + feed = store.feed_for(SPACE, "swe-worker") + assert [e["kind"] for e in feed] == ["item_transitioned", "item_commented"] + assert feed[0]["payload"]["comment"] == "totals drift" + assert feed[1]["payload"]["body"] == "use the v2 endpoint" + store.consume_feed(SPACE, "swe-worker", feed[-1]["seq"]) + # acceptance is in-slice too: the worker hears closure (and can pick up next) store.transition(SPACE, WORKER, item_id, "review", comment="fixed") store.transition(SPACE, LEAD, item_id, "done") - assert store.pending_for("swe-worker") == [] + assert [e["payload"]["to"] for e in store.feed_for(SPACE, "swe-worker")] == ["done"] -def test_cancel_notice_is_addressed_to_the_assignee(store): +def test_feed_reassignment_delivers_then_interest_ends(store): item_id = assigned(store) - store.consume("swe-worker", store.pending_for("swe-worker")[-1]["seq"]) + store.consume_feed(SPACE, "swe-worker", store.feed_for(SPACE, "swe-worker")[-1]["seq"]) + store.assign(SPACE, LEAD, item_id, "other-worker") + # the loser hears the reassignment… + feed = store.feed_for(SPACE, "swe-worker") + assert [e["kind"] for e in feed] == ["item_assigned"] + assert feed[0]["payload"]["previous"] == "swe-worker" + store.consume_feed(SPACE, "swe-worker", feed[-1]["seq"]) + # …then goes quiet: later events on the item never reach it + store.comment(SPACE, LEAD, item_id, "carry on") + assert store.feed_for(SPACE, "swe-worker") == [] + # while the new holder gets the item's whole story (its slice now) — the + # history it needs to pick the work up + kinds = [e["kind"] for e in store.feed_for(SPACE, "other-worker")] + assert kinds == [ + "item_created", + "item_assigned", + "item_assigned", + "item_commented", + ] + + +def test_cancel_reaches_the_assignee_through_the_feed(store): + item_id = assigned(store) + store.consume_feed(SPACE, "swe-worker", store.feed_for(SPACE, "swe-worker")[-1]["seq"]) store.transition(SPACE, LEAD, item_id, "canceled", comment="scope cut") - pending = store.pending_for("swe-worker") + pending = store.feed_for(SPACE, "swe-worker") assert len(pending) == 1 assert pending[0]["kind"] == "item_transitioned" assert pending[0]["payload"]["to"] == "canceled"