diff --git a/coworker/agent.py b/coworker/agent.py index 5456312d..a9f391ce 100644 --- a/coworker/agent.py +++ b/coworker/agent.py @@ -213,6 +213,7 @@ def build_engine( plan_approver: Optional[Any] = None, question_asker: Optional[Any] = None, tool_requester: Optional[Any] = None, + team_approver: Optional[Any] = None, subscription_store: Optional[Any] = None, channel_buffer: Optional[Any] = None, routing_targets: Optional[list[str]] = None, @@ -428,6 +429,13 @@ def build_engine( # call whenever the session isn't actually in plan mode. registry.register(propose_plan_tool()) + # The staffing gate — leads only. The engine intercepts it (out-of-band approval); + # approval pre-spawns the worker sessions and returns actor ids to assign to. + if agent.team == "lead": + from .teams.tools import propose_team_tool + + registry.register(propose_team_tool()) + # Per-turn ephemeral context, appended to the latest user message since mid-thread system # messages aren't reliable across providers. Three producers: the plan-mode reminder (mode can # flip mid-session, so it's checked each turn, not baked into the instructions), the live @@ -502,6 +510,7 @@ def build_engine( plan_approver=plan_approver, question_asker=question_asker, tool_requester=tool_requester, + team_approver=team_approver, ) engine.executor = executor # type: ignore[attr-defined] engine.todo = todo # type: ignore[attr-defined] diff --git a/coworker/agents/base.py b/coworker/agents/base.py index 0fb78cd1..bfd11b0b 100644 --- a/coworker/agents/base.py +++ b/coworker/agents/base.py @@ -41,6 +41,9 @@ class Agent: family: str = "knowledge" messaging: bool = False connectors: bool | tuple[str, ...] = False + # Team identity: "lead" | "worker" | None (solo-only). Gates the board/journal + # toolsets and staffing eligibility — solo personas are never team-staffable. + team: Optional[str] = None def build_tools(self, context: AgentContext) -> list: return list(self.tool_factory(context)) if self.tool_factory else [] diff --git a/coworker/conversations.py b/coworker/conversations.py index e67ef2fb..2915a3a9 100644 --- a/coworker/conversations.py +++ b/coworker/conversations.py @@ -96,6 +96,7 @@ class ConversationStore: "ALTER TABLE sessions ADD COLUMN renamed INTEGER DEFAULT 0", "ALTER TABLE sessions ADD COLUMN grants TEXT", "ALTER TABLE sessions ADD COLUMN compaction TEXT", + "ALTER TABLE sessions ADD COLUMN team TEXT", ): try: self._conn.execute(ddl) @@ -192,13 +193,14 @@ class ConversationStore: title = record.title or title_from(record.messages) self._conn.execute( """ - INSERT INTO sessions (session_id, workspace, model, mode, title, agent, n_msgs, messages, extra_roots, grants, compaction, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, CURRENT_TIMESTAMP) + INSERT INTO sessions (session_id, workspace, model, mode, title, agent, n_msgs, messages, extra_roots, grants, compaction, team, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, CURRENT_TIMESTAMP) ON CONFLICT(session_id) DO UPDATE SET workspace = excluded.workspace, model = excluded.model, mode = excluded.mode, title = COALESCE(sessions.title, excluded.title), agent = excluded.agent, n_msgs = excluded.n_msgs, messages = NULL, extra_roots = excluded.extra_roots, grants = excluded.grants, compaction = excluded.compaction, + team = excluded.team, updated_at = CURRENT_TIMESTAMP """, ( @@ -212,6 +214,7 @@ class ConversationStore: json.dumps(record.extra_roots or []), json.dumps(record.grants or {}), json.dumps(record.compaction or {}), + json.dumps(record.team or {}), ), ) self._conn.commit() @@ -252,6 +255,7 @@ class ConversationStore: archived=bool(row["archived"]), origin=row["origin"], origin_label=row["origin_label"], + team=_load_grants(row["team"] if "team" in row.keys() else None), ) def set_extra_roots(self, session_id: str, extra_roots: list[dict]) -> None: @@ -290,6 +294,7 @@ class ConversationStore: archived=bool(r["archived"]), origin=r["origin"], origin_label=r["origin_label"], + team=_load_grants(r["team"] if "team" in r.keys() else None), ) for r in rows ] diff --git a/coworker/engine.py b/coworker/engine.py index 613b94a6..c2ab0f02 100644 --- a/coworker/engine.py +++ b/coworker/engine.py @@ -89,6 +89,9 @@ class TurnEngine: tool_requester: Optional[ Callable[[dict[str, Any]], "Awaitable[dict[str, Any]]"] ] = None, + team_approver: Optional[ + Callable[[dict[str, Any]], "Awaitable[dict[str, Any]]"] + ] = None, # Called (thread-safe, best-effort) when the user stops the turn — e.g. the # executor's kill for a running shell command. interrupt_hooks: Optional[list[Callable[[], None]]] = None, @@ -119,6 +122,10 @@ class TurnEngine: # An approving result flips the live PermissionEngine out of plan mode (same session, # context kept). None on surfaces that can't prompt (the tool then no-ops). self.plan_approver = plan_approver + # Handles the `propose_team` tool (the staffing gate): emits TEAM_PROPOSED, waits + # for the user's decision; approval pre-spawns the worker sessions and the result + # carries the roster (actor ids). None on surfaces that can't prompt. + self.team_approver = team_approver # Handles the `ask_user` tool: turns a question into an Inbox item and waits for the answer # (answerable inline in a live session or from the Inbox when unattended). None on surfaces # that can't ask (the tool then no-ops). @@ -644,6 +651,10 @@ class TurnEngine: async for event in self._handle_plan_proposal(tool_call): yield event continue + if tool_call.name == "propose_team": + async for event in self._handle_team_proposal(tool_call): + yield event + continue if tool_call.name == "ask_user": async for event in self._handle_ask_user(tool_call): yield event @@ -919,6 +930,56 @@ class TurnEngine: except Exception: pass + async def _handle_team_proposal(self, tool_call: ToolCall) -> AsyncIterator[Event]: + """The staffing gate: emit the proposed roster, await the user's out-of-band + decision. Approval PRE-SPAWNS the worker sessions (server-side, inside the + approver) and the result carries the roster with actor ids so the lead can + assign; rejection returns the user's feedback for a revised proposal.""" + args = tool_call.arguments or {} + members = args.get("members") or [] + if not isinstance(members, list) or not members: + result: dict[str, Any] = { + "approved": False, + "error": "propose at least one member ({persona, model?, reason?})", + } + elif self.team_approver is None: + result = { + "approved": False, + "error": "team staffing isn't available in this surface", + } + else: + yield Event( + EventType.TEAM_PROPOSED, + { + "members": members, + "enable_chat": bool(args.get("enable_chat", False)), + "note": str(args.get("note", "")), + }, + ) + self._audit(tool_call, stage="team_proposed") + result = await self._interruptible( + self.team_approver(dict(args), tool_call.id), + interrupted={"approved": False, "error": "interrupted by user"}, + ) or {"approved": False, "error": "no response"} + + status = "ok" if result.get("approved") else "denied" + self.messages.append(_tool_result_message(tool_call, result)) + self._audit( + tool_call, + stage="finished", + status=status, + result=result, + result_preview=_preview(result), + ) + yield Event( + EventType.TOOL_FINISHED, + { + "name": tool_call.name, + "status": status, + "result_preview": _preview(result), + }, + ) + async def _handle_plan_proposal(self, tool_call: ToolCall) -> AsyncIterator[Event]: """Emit the plan for review, await the user's out-of-band decision, and apply it: approval flips the live PermissionEngine out of plan mode (the same session keeps diff --git a/coworker/events.py b/coworker/events.py index 457e1da4..1934436f 100644 --- a/coworker/events.py +++ b/coworker/events.py @@ -26,6 +26,9 @@ class EventType(str, Enum): PLAN_PROPOSED = ( "plan_proposed" # agent presents a plan for approval (plan mode exit) ) + TEAM_PROPOSED = ( + "team_proposed" # a lead proposes a worker roster (the staffing gate) + ) TOOL_STARTED = "tool_started" TOOL_FINISHED = "tool_finished" ITERATION_END = "iteration_end" diff --git a/coworker/personas/loading.py b/coworker/personas/loading.py index 27d3303e..f7478353 100644 --- a/coworker/personas/loading.py +++ b/coworker/personas/loading.py @@ -31,6 +31,9 @@ def consent_summary(m: PersonaManifest) -> dict: "connectors": "all" if m.connectors is True else list(m.connectors or ()), "mcp": list(m.mcp), "messaging": m.messaging, + # "lead" personas can create and direct worker coworkers — the consent + # screen says that plainly (capability firebreak as a manifest fact). + "team": m.team, "recommended_mode": m.default_permission_mode, "recommended_models": list(m.recommended_models), # Recommended connectors/MCP with reasons + tiers — the consent screen shows @@ -59,6 +62,10 @@ def capability_set(m: PersonaManifest) -> set[str]: caps |= {f"connector:{c}" for c in m.connectors or ()} if m.messaging: caps.add("messaging") + # An update that turns a solo persona into a lead/worker must re-consent — + # team capability changes who the coworker can direct or be directed by. + if m.team: + caps.add(f"team:{m.team}") return caps diff --git a/coworker/personas/manifest.py b/coworker/personas/manifest.py index 0ddeb517..d3aa47f7 100644 --- a/coworker/personas/manifest.py +++ b/coworker/personas/manifest.py @@ -21,6 +21,7 @@ import yaml _ID_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$") VALID_FAMILIES = {"code", "knowledge"} +VALID_TEAM = {"lead", "worker"} VALID_WORKSPACES = {"git", "project", "deliverable", "none"} VALID_MODES = {"discuss", "plan", "interactive", "custom", "auto"} VALID_REC_KINDS = {"connector", "mcp"} @@ -63,6 +64,13 @@ class PersonaManifest: # `all` sentinel, reserved for built-in general personas. Coarser grants leaked # undeclared tools (browser, email) into security sessions; undeclared = absent. connectors: bool | tuple[str, ...] = False + # Team identity (agent-teams design, third/fourth pass): "lead" = coordinates a + # team (gets the board coordination verbs + gates; consent copy says "can create + # and direct worker coworkers"); "worker" = purpose-built to work under a lead + # (board worker verbs, no ask_user-shaped prompt); None = solo-only. Solo + # personas are NOT team-eligible — team-awareness changes who the prompt talks + # to, so staffing fails closed on personas without the trait. + team: Optional[str] = None default_permission_mode: str = "interactive" recommended_models: list[str] = field(default_factory=list) skills: list[str] = field(default_factory=list) @@ -97,6 +105,7 @@ class PersonaManifest: family=self.family, messaging=self.messaging, connectors=self.connectors, + team=self.team, ) @@ -279,6 +288,13 @@ def parse_manifest( f"persona {persona_id!r}: default_permission_mode must be one of {sorted(VALID_MODES)}" ) + team_raw = str(meta.get("team", "") or "").strip().lower() + if team_raw and team_raw not in VALID_TEAM: + raise ManifestError( + f"persona {persona_id!r}: team must be one of {sorted(VALID_TEAM)}" + " (omit for a solo coworker)" + ) + tools = _strlist(meta, "tools") _validate_tools(persona_id, tools) recommends = _recommends(persona_id, meta) @@ -296,6 +312,7 @@ def parse_manifest( workspace=workspace, messaging=bool(meta.get("messaging", False)), connectors=connectors, + team=team_raw or None, default_permission_mode=mode, recommended_models=_strlist(meta, "recommended_models"), skills=_strlist(meta, "skills"), diff --git a/coworker/server/app.py b/coworker/server/app.py index e0e1b276..e2807b89 100644 --- a/coworker/server/app.py +++ b/coworker/server/app.py @@ -1835,6 +1835,43 @@ def create_app(manager: SessionManager) -> FastAPI: } return {"approved": True, "mode": resp.get("mode") or "interactive"} + async def team_approver(_args: dict, tool_call_id=None) -> dict: + # The staffing gate. The engine already emitted TEAM_PROPOSED; park an + # Inbox item as the durable resolution vehicle, wait for the verdict, and + # on approval PRE-SPAWN the team (create_team fails closed on non-worker + # personas, so a bad roster reads as a rejection with the reason). + members = _args.get("members") or [] + roster = "\n".join( + f"- {m.get('persona', '?')}" + + (f" · {m['model']}" if m.get("model") else "") + + (f" — {m['reason']}" if m.get("reason") else "") + for m in members + if isinstance(m, dict) + ) + item = manager.inbox.add_plan( + session_id, + "Create this team?", + body=roster, + inbox=_route(), + visibility=_visibility(), + tool_call_id=tool_call_id, + ) + if item.state == "pending": + manager.persist_session(session_id) + if item.visibility == VIS_INBOX: + await _mirror(item) + resp = _parse_json(await manager.inbox.wait(item.id)) + if not resp.get("approved"): + return { + "approved": False, + "feedback": resp.get("feedback") or "the user declined this roster", + } + return manager.create_team( + session_id, + [m for m in members if isinstance(m, dict)], + enable_chat=bool(_args.get("enable_chat", False)), + ) + async def _apply_model(model: Optional[str]) -> None: # Mid-session rebind is allowed (roadmap item 3, supersedes the 2026-07-04 # lock): history is canonical and providers convert per call. A real switch @@ -1874,6 +1911,7 @@ def create_app(manager: SessionManager) -> FastAPI: plan_approver=plan_approver, question_asker=question_asker, tool_requester=tool_requester, + team_approver=team_approver, ) if engine is None: await ws.send_json( @@ -2024,6 +2062,15 @@ def create_app(manager: SessionManager) -> FastAPI: } ) ) + elif kind == "team_response": + _resolve_pending( + json.dumps( + { + "approved": bool(message.get("approved")), + "feedback": message.get("feedback", ""), + } + ) + ) elif kind == "question_response": _resolve_pending(str(message.get("answer", ""))) elif kind == "interrupt": diff --git a/coworker/server/manager.py b/coworker/server/manager.py index f54336df..dc3611c6 100644 --- a/coworker/server/manager.py +++ b/coworker/server/manager.py @@ -15,6 +15,7 @@ import re import shutil import subprocess import time +import uuid from pathlib import Path from typing import Any, Optional @@ -86,6 +87,7 @@ from ..teams import Actor as TeamActor from ..teams import BoardError as TeamsBoardError from ..teams import JournalStore, Role as TeamRole, TeamStore, board_tools, journal_tools from ..teams.model import space_for_workspace +from ..teams.registry import TeamRegistry, TeamWorker from ..skills import ( SessionSkillStore, SkillLoader, @@ -198,14 +200,18 @@ class SessionManager: # The scheduler also resumes self-wake'd sessions each tick (extra_tick). self.task_store = TaskStore(base / "automation.db") self.scheduler = Scheduler( - self.task_store, self._run_scheduled_task, extra_tick=self.resume_due_wakes + self.task_store, self._run_scheduled_task, extra_tick=self._scheduler_tick ) # Agent teams: two append-only stores, one record discipline. The journal is # case-keyed (knowledge outlives boards/teams); the board log is space-scoped, # and assignment feeds journal-case grants. Verbs register per-session behind - # the persona's `team:` trait (wake plumbing lands separately). + # the persona's `team:` trait; the registry holds rosters (lead/worker + # sessions per board) that the wake plumbing walks. self.journal_store = JournalStore(base / "journal.db") self.team_store = TeamStore(base / "teams.db", journal=self.journal_store) + self.teams = TeamRegistry(base / "teams.json") + self._team_inflight: set[str] = set() + self._loop: Optional[asyncio.AbstractEventLoop] = None # Personas: registry + lifecycle state under this manager's data dir. Installed as the # process singleton so agents.get_agent resolves persona ids (incl. third-party) here. self.personas = PersonaRegistry(state_path=base / "personas.json") @@ -477,6 +483,7 @@ class SessionManager: plan_approver: Optional[Any] = None, question_asker: Optional[Any] = None, tool_requester: Optional[Any] = None, + team_approver: Optional[Any] = None, ) -> Optional[TurnEngine]: engine = self._engines.get(session_id) if engine is not None: @@ -490,6 +497,8 @@ class SessionManager: engine.question_asker = question_asker if tool_requester is not None: engine.tool_requester = tool_requester + if team_approver is not None: + engine.team_approver = team_approver return engine record = self.session_store.load(session_id) @@ -546,7 +555,7 @@ class SessionManager: messages=messages, extra_tools=[ *(extra_tools or []), - *self._team_board_tools(session_id, agent_name, ws), + *self._team_tools_for(session_id, ag, record, ws), ] or None, secrets=self.secrets, @@ -566,6 +575,7 @@ class SessionManager: question_asker=question_asker or self.inbox_question_asker(session_id, agent), tool_requester=tool_requester, + team_approver=team_approver, subscription_store=self.subscriptions, channel_buffer=self.channel_buffer, routing_targets=self._routing_targets(session_id, agent), @@ -1415,23 +1425,308 @@ class SessionManager: def journal_overview(self) -> list[dict[str, Any]]: return self.journal_store.overview(self._user_actor()) - def _team_board_tools(self, session_id: str, agent_name: str, ws: Optional[str]) -> list[Any]: - """Phase-1 experimental wiring (flag: OPENWORKER_TEAM_BOARD=1): every - workspace session gets the board+journal verbs as the LEAD of its - workspace's board. Registration moves behind the persona `team:` trait - with the wake plumbing.""" - if not ws or os.environ.get("OPENWORKER_TEAM_BOARD") != "1": + TEAM_WAKE_CAP_PER_HOUR = 60 # budget gate at the wake gate: silent server cap + + def _team_tools_for( + self, session_id: str, agent: Any, record: Any, ws: Optional[str] + ) -> list[Any]: + """Board/journal verbs, gated by the persona `team:` trait. Leads get the + coordination set (+ steer); workers get the worker set bound to their roster + actor id. OPENWORKER_TEAM_BOARD=1 keeps the phase-1 any-session-as-lead dev + mode.""" + role = getattr(agent, "team", None) + if role is None and ws and os.environ.get("OPENWORKER_TEAM_BOARD") == "1": + role = "lead" + if role is None or not ws: return [] - actor = TeamActor( - id=f"{agent_name}:{session_id[:8]}", - role=TeamRole.LEAD, - persona=agent_name, - session_id=session_id, - ) space = space_for_workspace(ws) - return board_tools(self.team_store, space=space, actor=actor) + journal_tools( + if role == "worker": + info = (record.team if record is not None else {}) or {} + actor = TeamActor( + id=str(info.get("actor") or f"{agent.name}:{session_id[:8]}"), + role=TeamRole.WORKER, + persona=agent.name, + session_id=session_id, + ) + space = str(info.get("space") or space) + else: + actor = TeamActor( + id=f"{agent.name}:{session_id[:8]}", + role=TeamRole.LEAD, + persona=agent.name, + session_id=session_id, + ) + tools = board_tools(self.team_store, space=space, actor=actor) + journal_tools( self.journal_store, actor=actor, space=space ) + if role == "lead": + tools.append(self._steer_tool(session_id)) + return tools + + def _steer_tool(self, lead_session_id: str) -> Any: + """The lead's downward steering verb. Text lands in the worker's session + attributed [Lead] — queued into a live turn, or a fresh background turn when + idle. Strictly downward: no worker ever gets this tool.""" + import aisuite as ai + + manager = self + + def steer_worker(worker: str, message: str) -> dict: + """Send steering text to one of your workers (by actor id). Use for + exceptions — changed requirements, stop/redirect, unblock guidance; + routine status flows through the board, not steering.""" + team = manager.teams.for_lead_session(lead_session_id) + if team is None: + return {"error": "no team yet — propose one with propose_team first"} + match = next((w for w in team.workers if w.actor == worker), None) + if match is None: + return { + "error": f"no worker '{worker}' on this team", + "workers": [w.actor for w in team.workers], + } + if manager._loop is None: + return {"error": "steering is unavailable in this surface"} + asyncio.run_coroutine_threadsafe( + manager.deliver_to_session( + match.session_id, f"[Lead] {message}".strip() + ), + manager._loop, + ) + return {"ok": True, "delivered_to": worker} + + return ai.tool( + steer_worker, + metadata=ai.ToolMetadata( + category="team", risk_level="medium", capabilities=["team"] + ), + ) + + def create_team( + self, session_id: str, members: list[dict[str, Any]], *, enable_chat: bool = False + ) -> dict[str, Any]: + """The staffing gate's approved action: PRE-SPAWN worker sessions (state on + disk, zero tokens — the first model turn fires when the first assignment + lands) and register the team. Fails closed on personas without `team: worker`.""" + record = self.session_store.load(session_id) + if record is None or not record.workspace: + return {"approved": False, "error": "the lead session has no workspace"} + if self.teams.for_lead_session(session_id) is not None: + return {"approved": False, "error": "this session already leads a team"} + space = space_for_workspace(record.workspace) + workers: list[TeamWorker] = [] + used: set[str] = set() + for member in members: + pid = str((member or {}).get("persona", "")).strip() + try: + ag = get_agent(pid) + except Exception: + return { + "approved": False, + "error": f"unknown coworker '{pid}' — it must be installed and enabled", + } + if getattr(ag, "team", None) != "worker": + # Fail closed: solo personas are not team-eligible — their prompts + # are written at a human, not a lead. + return { + "approved": False, + "error": f"'{pid}' is not team-capable (needs `team: worker`)", + } + actor, n = pid, 2 + while actor in used: + actor, n = f"{pid}-{n}", n + 1 + used.add(actor) + worker_sid = uuid.uuid4().hex[:12] + model = str(member.get("model") or record.model) + self.session_store.save( + SessionRecord( + session_id=worker_sid, + workspace=record.workspace, + model=model, + mode=record.mode, + messages=[], + agent=pid, + team={ + "role": "worker", + "actor": actor, + "lead_session": session_id, + "space": space, + }, + ) + ) + workers.append( + TeamWorker(actor=actor, persona=pid, session_id=worker_sid, model=model) + ) + team = self.teams.create( + space=space, + lead_session=session_id, + lead_actor=f"{record.agent}:{session_id[:8]}", + workers=workers, + chat_enabled=enable_chat, + ) + for worker in workers: + self._emit_session_created(worker.session_id, worker.persona) + record.team = { + "team_id": team.team_id, + "role": "lead", + "actor": team.lead_actor, + "space": space, + } + self.session_store.save(record) + return { + "approved": True, + "team_id": team.team_id, + "workers": [ + {"actor": w.actor, "persona": w.persona, "session_id": w.session_id} + for w in workers + ], + "note": ( + "team created — workers are idle until you assign. Create work items" + " and assign them to the actor ids above; review-state items are" + " yours to verify." + ), + } + + async def team_tick(self) -> int: + """Drain team queues (called each scheduler tick + kicked after team turns). + One wake consumes a burst as one digest; durable-until-consumed — the cursor + advances only after the delivery turn is dispatched.""" + delivered = 0 + for team in self.teams.all(): + if team.paused: + continue + for worker in team.workers: + delivered += await self._drain_team_member( + team, + session_id=worker.session_id, + actor=worker.actor, + is_lead=False, + ) + delivered += await self._drain_team_member( + team, + session_id=team.lead_session, + actor=team.lead_actor, + is_lead=True, + ) + return delivered + + async def _drain_team_member( + self, team, *, session_id: str, actor: str, is_lead: bool + ) -> int: + directs = self.team_store.pending_for(actor) + subs = ( + self.team_store.subscribed_events(team.space, actor) if is_lead else [] + ) + if not directs and not subs: + return 0 + if self.is_running(session_id) or session_id in self._team_inflight: + return 0 # it will drain on its next turn end / next tick + 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 = self._team_digest(team, directs, subs, is_lead=is_lead) + self._team_inflight.add(session_id) + + async def _deliver() -> None: + try: + await self.deliver_to_session(session_id, message) + # 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"]) + if subs: + self.team_store.consume_subscription( + team.space, actor, subs[-1]["seq"] + ) + finally: + self._team_inflight.discard(session_id) + + asyncio.create_task(_deliver()) + return 1 + + def _team_digest( + self, team, directs: list[dict], subs: list[dict], *, is_lead: bool + ) -> str: + """Coalesce one queue batch into one wake message. Deterministic, computed + by code — the model does judgment, not arithmetic.""" + lines: list[str] = [] + for event in directs + subs: + item_id = event.get("item_id") + payload = event.get("payload") or {} + item = None + if item_id is not None: + try: + item = self.team_store.get_item(team.space, int(item_id)) + except Exception: + item = None + title = f"#{item_id} {item['title']}" if item else f"#{item_id}" + if event["kind"] == "item_assigned": + if item is None: + 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 "") + ) + elif event["kind"] == "item_transitioned": + to = payload.get("to", "?") + note = f" — “{payload.get('comment')}”" if payload.get("comment") else "" + lines.append(f"{title} moved to {to} by {event['actor']}{note}") + elif event["kind"] == "item_created": + lines.append(f"New item filed by {event['actor']}: {title}") + elif event["kind"] == "item_commented": + lines.append( + f"Comment on {title} by {event['actor']}: {payload.get('body', '')}" + ) + body = "\n".join(f"- {line}" for line in lines) or "- (no detail)" + if is_lead: + return ( + "⏰ Board wake — your team needs decisions:\n" + + body + + "\n\nVerify review items against their acceptance criteria (then" + " done, or send back with a comment), unblock or reassign blocked" + " items, and triage new filings. Steer only where needed." + ) + return ( + "[Lead] Board update:\n" + + body + + "\n\nMove your item to in_progress when you start; blocked (with a" + " comment) if stuck; review with a hand-off comment when finished." + " Journal evidence as you go." + ) + + def team_staleness_digest(self, session_id: str) -> str: + """Attached to a lead's TIMER wakes: pure code over the board — a + nothing's-wrong wake is one cheap glance, never a re-survey. Scoped by + role membership: sessions with no team role get nothing.""" + team = self.teams.for_lead_session(session_id) + if team is None: + return "" + try: + items = self.team_store.list_items(team.space, self._user_actor()) + except Exception: + return "" + by_state: dict[str, int] = {} + for item in items: + by_state[item["state"]] = by_state.get(item["state"], 0) + 1 + unassigned = sum( + 1 for i in items if i["state"] == "open" and not i["assignee"] + ) + parts = [f"{n} {state}" for state, n in sorted(by_state.items())] + lines = [f"Board: {', '.join(parts) or 'empty'}."] + if unassigned: + lines.append(f"{unassigned} open item(s) have no assignee.") + reviews = [i for i in items if i["state"] == "review"] + if reviews: + lines.append( + "Awaiting your review: " + + ", ".join(f"#{i['id']} {i['title']}" for i in reviews[:5]) + ) + blocked = [i for i in items if i["state"] == "blocked"] + if blocked: + lines.append( + "Blocked: " + ", ".join(f"#{i['id']} {i['title']}" for i in blocked[:5]) + ) + return "\n".join(lines) def list_artifacts(self, session_id: str) -> list[dict[str, Any]]: record = self.session_store.load(session_id) @@ -2615,6 +2910,8 @@ class SessionManager: """Build the messaging gateway and start enabled listeners. Inbound messages route to durable sessions: a channel message to its subscribers, a DM to the designated DM session (else parked). Returns the platforms whose listeners came up.""" + # Team steering/kicks are dispatched from tool threads; they need the app loop. + self._loop = asyncio.get_running_loop() self.scheduler.start() # tick scheduler for automations (independent of connectors) return await self._build_and_start_gateway() @@ -3069,6 +3366,16 @@ class SessionManager: return resolve_from_reply(text, _resolve) is not None # -- self-wake resumption --------------------------------------------------- + async def _scheduler_tick(self) -> None: + """The shared per-tick work: resume due self-wakes, then drain team queues. + Team deliveries dispatch as tasks (a long worker turn must not stall the + scheduler).""" + await self.resume_due_wakes() + try: + await self.team_tick() + except Exception: + logger.exception("team tick failed") + async def resume_due_wakes(self) -> int: """Resume sessions whose self-wakes are due (called each scheduler tick). A suspended agent (it called sleep_for / wake_on / wake_on_event and ended its turn) is re-invoked on @@ -3101,12 +3408,26 @@ class SessionManager: # finishes — the one shared post-turn moment, so auto-titling hooks in here and # can never add latency to the response itself. self._maybe_autotitle(session_id) + # Team sessions: a finished turn is the moment new board events exist (an + # assign, a review transition) — kick the queue drain now instead of waiting + # for the next scheduler tick. Cheap no-op for teamless sessions. + if self._loop is not None and ( + self.teams.for_lead_session(session_id) + or self.teams.for_worker_session(session_id) + ): + asyncio.run_coroutine_threadsafe(self.team_tick(), self._loop) def is_running(self, session_id: str) -> bool: return session_id in self._running_sessions async def _resume_wake(self, wake) -> None: - await self.deliver_to_session(wake.session_id, self._wake_message(wake)) + message = self._wake_message(wake) + # A lead's timer wake carries the staleness digest — pure code over the + # board, scoped by role membership (teamless sessions get a bare wake). + digest = self.team_staleness_digest(wake.session_id) + if digest: + message = f"{message}\n\n{digest}" + await self.deliver_to_session(wake.session_id, message) async def deliver_to_session( self, session_id: str, message: str, *, source: Optional[dict[str, Any]] = None @@ -4004,11 +4325,47 @@ class SessionManager: "subscriptions": [ s.channel for s in self.subscriptions.for_session(r.session_id) ], + # Agent teams: {} for plain sessions. Workers carry role/lead_session + # (+ a computed current-item line); leads carry role/team_id — drives + # the sidebar's ONE expandable team entry. + "team": self._session_team_row(r), } for r in self.session_store.list(workspace=ws) if not r.session_id.startswith("__") # hide internal threads ] + def _session_team_row(self, record: SessionRecord) -> dict[str, Any]: + info = record.team or {} + if not info: + return {} + row = { + "role": info.get("role", ""), + "team_id": info.get("team_id", ""), + "lead_session": info.get("lead_session", ""), + } + if info.get("role") == "worker" and info.get("space") and info.get("actor"): + try: + items = self.team_store.list_items( + str(info["space"]), self._user_actor(), assignee=str(info["actor"]) + ) + except Exception: + items = [] + active = next( + ( + i + for state in ("blocked", "review", "in_progress", "open") + for i in items + if i["state"] == state + ), + None, + ) + row["actor"] = info["actor"] + row["current_item"] = ( + f"#{active['id']} {active['state'].replace('_', ' ')}" if active else "idle" + ) + row["status"] = active["state"] if active else "idle" + return row + def _session_liveness(self, session_id: str) -> str: if self.is_running(session_id): return "working" diff --git a/coworker/sessions.py b/coworker/sessions.py index 4bd857c7..33f176d5 100644 --- a/coworker/sessions.py +++ b/coworker/sessions.py @@ -37,3 +37,7 @@ class SessionRecord: # Auto-compaction state (OPE-27): CompactionState.as_dict(), {} when never compacted. # Persisted so a reloaded session keeps its compacted outbound view. compaction: dict[str, Any] = field(default_factory=dict) + # Agent teams: {} for plain sessions. Workers: {team_id, role: "worker", actor, + # lead_session, space}. Leads gain their entry when the staffing gate creates the + # team. Drives tool binding (board actor identity) + the sidebar's expandable entry. + team: dict[str, Any] = field(default_factory=dict) diff --git a/coworker/teams/registry.py b/coworker/teams/registry.py new file mode 100644 index 00000000..53d7a855 --- /dev/null +++ b/coworker/teams/registry.py @@ -0,0 +1,136 @@ +"""Team registry — which sessions form a team: one lead, its workers, their board. + +A team is created at the staffing gate ("Create team & start"): worker sessions are +PRE-SPAWNED as durable state on disk (spawn ≠ first turn — an unassigned worker costs +zero tokens; its first model turn fires when the first assignment lands). The registry +is the roster the wake plumbing walks each tick, and the tie that scopes staleness +digests by role membership. +""" + +from __future__ import annotations + +import json +import threading +import uuid +from dataclasses import asdict, dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Optional + + +@dataclass +class TeamWorker: + actor: str # board actor id (stable; assignments address this) + persona: str + session_id: str + model: str = "" + + +@dataclass +class Team: + team_id: str + space: str + lead_session: str + lead_actor: str + workers: list[TeamWorker] = field(default_factory=list) + chat_enabled: bool = False + paused: bool = False # budget/user pause: the wake gate skips a paused team + created_at: str = field( + default_factory=lambda: datetime.now(timezone.utc).isoformat() + ) + # Rolling budget gate: automatic wakes this hour (reset when the hour rolls). + wake_hour: str = "" + wakes_this_hour: int = 0 + + +class TeamRegistry: + def __init__(self, path: Optional[str | Path] = None) -> None: + self.path = Path(path) if path else None + self._lock = threading.Lock() + self._teams: dict[str, Team] = {} + if self.path and self.path.is_file(): + for raw in json.loads(self.path.read_text(encoding="utf-8")).get( + "teams", [] + ): + workers = [TeamWorker(**w) for w in raw.pop("workers", [])] + team = Team(**{**raw, "workers": []}) + team.workers = workers + self._teams[team.team_id] = team + + def _save(self) -> None: + if not self.path: + return + self.path.parent.mkdir(parents=True, exist_ok=True) + self.path.write_text( + json.dumps( + {"teams": [asdict(t) for t in self._teams.values()]}, indent=2 + ), + encoding="utf-8", + ) + + def create( + self, + *, + space: str, + lead_session: str, + lead_actor: str, + workers: list[TeamWorker], + chat_enabled: bool = False, + ) -> Team: + team = Team( + team_id=uuid.uuid4().hex[:12], + space=space, + lead_session=lead_session, + lead_actor=lead_actor, + workers=workers, + chat_enabled=chat_enabled, + ) + with self._lock: + self._teams[team.team_id] = team + self._save() + return team + + def all(self) -> list[Team]: + return list(self._teams.values()) + + def get(self, team_id: str) -> Optional[Team]: + return self._teams.get(team_id) + + def for_lead_session(self, session_id: str) -> Optional[Team]: + for team in self._teams.values(): + if team.lead_session == session_id: + return team + return None + + def for_worker_session(self, session_id: str) -> Optional[tuple[Team, TeamWorker]]: + for team in self._teams.values(): + for worker in team.workers: + if worker.session_id == session_id: + return team, worker + return None + + def set_paused(self, team_id: str, paused: bool) -> None: + with self._lock: + team = self._teams.get(team_id) + if team is not None: + team.paused = paused + self._save() + + def count_wake(self, team_id: str, *, cap: int) -> bool: + """The budget gate at the wake gate: count one automatic wake against the + team's rolling hour; False = over cap (the caller skips the wake and the + team reads as paused-for-budget until the hour rolls). A runaway loop + stops BETWEEN turns, never mid-flight.""" + hour = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H") + with self._lock: + team = self._teams.get(team_id) + if team is None: + return False + if team.wake_hour != hour: + team.wake_hour, team.wakes_this_hour = hour, 0 + if team.wakes_this_hour >= cap: + self._save() + return False + team.wakes_this_hour += 1 + self._save() + return True diff --git a/coworker/teams/store.py b/coworker/teams/store.py index 9e413d78..4599c02e 100644 --- a/coworker/teams/store.py +++ b/coworker/teams/store.py @@ -133,6 +133,10 @@ class TeamStore: head_hash TEXT NOT NULL, watermark INTEGER NOT NULL ); + CREATE TABLE IF NOT EXISTS team_cursors ( + cursor_key TEXT PRIMARY KEY, + consumed_seq INTEGER NOT NULL + ); """) self._conn.commit() @@ -276,6 +280,70 @@ class TeamStore: ).fetchall() return [_row_to_event(row) for row in rows] + # -------------------------------------------------- delivery (durable queue) + + # 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. + + 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 consume(self, recipient: str, upto_seq: int) -> None: + self._set_cursor(f"to:{recipient}", 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 + # appends and routine comments never wake anyone. + SUBSCRIBED_TRANSITIONS = ("review", "blocked") + + def subscribed_events( + self, space: str, subscriber: str, *, limit: int = 200 + ) -> list[dict[str, Any]]: + """Unconsumed subscription-worthy events on a space for one subscriber.""" + key = f"sub:{subscriber}:{space}" + events = self.events( + space, + kinds=[ITEM_TRANSITIONED, ITEM_CREATED], + since_seq=self._cursor(key), + limit=limit, + ) + out = [] + for event in events: + if event["actor"] == subscriber: + continue # your own verbs never wake you + if ( + event["kind"] == ITEM_TRANSITIONED + and event["payload"].get("to") not in self.SUBSCRIBED_TRANSITIONS + ): + continue + out.append(event) + return out + + def consume_subscription(self, space: str, subscriber: str, upto_seq: int) -> None: + self._set_cursor(f"sub:{subscriber}:{space}", upto_seq) + + def _cursor(self, key: str) -> int: + row = self._conn.execute( + "SELECT consumed_seq FROM team_cursors WHERE cursor_key = ?", (key,) + ).fetchone() + return int(row["consumed_seq"]) if row else 0 + + def _set_cursor(self, key: str, seq: int) -> None: + with self._lock: + self._conn.execute( + "INSERT INTO team_cursors (cursor_key, consumed_seq) VALUES (?, ?)" + " ON CONFLICT(cursor_key) DO UPDATE SET consumed_seq =" + " MAX(consumed_seq, ?)", + (key, int(seq), int(seq)), + ) + self._conn.commit() + def spaces(self) -> list[str]: with self._lock: rows = self._conn.execute( diff --git a/coworker/teams/tools.py b/coworker/teams/tools.py index fe22a209..52893c7a 100644 --- a/coworker/teams/tools.py +++ b/coworker/teams/tools.py @@ -218,6 +218,68 @@ def journal_tools( return [_wrap(local[name]) for name in JOURNAL_VERBS] +# The staffing gate's schema carrier. Like propose_plan, the real handling lives in +# the TurnEngine (it needs the out-of-band approval round-trip): it emits +# TEAM_PROPOSED and waits; approval PRE-SPAWNS the worker sessions and returns the +# roster (actor ids) to the lead. This body only runs when no approver is wired. +_PROPOSE_TEAM_SCHEMA = { + "type": "function", + "function": { + "name": "propose_team", + "description": ( + "Propose the worker coworkers you need for this board. The user sees the" + " roster and approves it; approval creates the worker sessions and" + " returns their actor ids so you can assign work items to them. Only" + " team-capable worker coworkers may be proposed." + ), + "parameters": { + "type": "object", + "properties": { + "members": { + "type": "array", + "items": { + "type": "object", + "properties": { + "persona": {"type": "string"}, + "model": {"type": "string"}, + "reason": {"type": "string"}, + }, + "required": ["persona"], + }, + }, + "enable_chat": {"type": "boolean"}, + "note": {"type": "string"}, + }, + "required": ["members"], + }, + }, +} + + +def propose_team_tool() -> object: + def propose_team( + members: Optional[list] = None, enable_chat: bool = False, note: str = "" + ) -> dict: + """Propose the worker roster for this board (the staffing gate). Each member + is {persona, model?, reason?}. The user approves; approval creates the + worker sessions and returns their actor ids for assignment.""" + return { + "approved": False, + "error": "team staffing isn't available in this surface", + } + + wrapped = ai.tool( + propose_team, + metadata=ai.ToolMetadata( + category="team", + risk_level="medium", + capabilities=["team"], + ), + ) + wrapped.__coworker_schema__ = _PROPOSE_TEAM_SCHEMA + return wrapped + + def _call(func, *args, **kwargs) -> dict: try: result = func(*args, **kwargs) diff --git a/tests/test_team_wake.py b/tests/test_team_wake.py new file mode 100644 index 00000000..64340f8d --- /dev/null +++ b/tests/test_team_wake.py @@ -0,0 +1,225 @@ +"""OPE-97 wake plumbing: the team trait, delivery cursors, lead subscriptions, +the registry + budget gate, pre-spawn at staffing, and digests.""" + +import pytest + +from coworker.personas.loading import capability_set +from coworker.personas.manifest import ManifestError, parse_manifest +from coworker.server.manager import SessionManager +from coworker.teams import Actor, Role, TeamStore +from coworker.teams.registry import TeamRegistry, TeamWorker + +USER = Actor(id="user", role=Role.USER) +LEAD = Actor(id="lead-1", role=Role.LEAD) +WORKER = Actor(id="swe-worker", role=Role.WORKER) +SPACE = "proj" + + +def manifest(team_line=""): + return f"""--- +id: t +name: T +family: code +tools: [search] +{team_line} +--- +Prompt body. +""" + + +# ------------------------------------------------------------------- team trait + +def test_team_trait_parses_and_gates_capabilities(): + lead = parse_manifest(manifest("team: lead")) + worker = parse_manifest(manifest("team: worker")) + solo = parse_manifest(manifest()) + assert lead.team == "lead" and worker.team == "worker" and solo.team is None + assert "team:lead" in capability_set(lead) + assert "team:worker" in capability_set(worker) + assert not any(c.startswith("team:") for c in capability_set(solo)) + # the trait reaches the runtime Agent (it gates tool registration) + assert lead.to_agent().team == "lead" + + +def test_invalid_team_trait_fails_loudly(): + with pytest.raises(ManifestError, match="team"): + parse_manifest(manifest("team: manager")) + + +# ------------------------------------------------------- delivery cursors/queue + +@pytest.fixture +def store(tmp_path): + store = TeamStore(tmp_path / "teams.db") + yield store + store.close() + + +def assigned(store, assignee="swe-worker"): + item = store.create_item(SPACE, LEAD, title="Task", criteria="tests pass") + store.assign(SPACE, LEAD, item["id"], assignee) + return item["id"] + + +def test_deliveries_are_durable_until_consumed(store): + assigned(store) + first = store.pending_for("swe-worker") + assert len(first) == 1 and first[0]["kind"] == "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") == [] + # a second assignment queues fresh + assigned(store) + assert len(store.pending_for("swe-worker")) == 1 + + +def test_lead_subscriptions_are_an_allowlist(store): + item_id = assigned(store) + store.transition(SPACE, WORKER, item_id, "in_progress") # not subscribed + store.comment(SPACE, WORKER, item_id, "halfway") # never wakes + store.transition(SPACE, WORKER, item_id, "review", comment="done, please check") + filed = store.create_item(SPACE, WORKER, title="Found a bug", criteria="fix") + subs = store.subscribed_events(SPACE, "lead-1") + # Exactly the worker's review transition + the worker's filing: the lead's own + # verbs, the in_progress transition, and the comment never wake it. + assert all(e["actor"] != "lead-1" for e in subs) + assert {(e["kind"], e["payload"].get("to")) for e in subs} == { + ("item_transitioned", "review"), + ("item_created", None), + } + store.consume_subscription(SPACE, "lead-1", subs[-1]["seq"]) + assert store.subscribed_events(SPACE, "lead-1") == [] + _ = filed + + +# --------------------------------------------------------------- registry/budget + +def test_registry_roundtrip_and_budget_cap(tmp_path): + path = tmp_path / "teams.json" + reg = TeamRegistry(path) + team = reg.create( + space=SPACE, + lead_session="lead-sid", + lead_actor="lead-1", + workers=[TeamWorker(actor="swe-worker", persona="swe-worker", session_id="w1")], + ) + again = TeamRegistry(path) + loaded = again.get(team.team_id) + assert loaded is not None and loaded.workers[0].session_id == "w1" + assert again.for_lead_session("lead-sid").team_id == team.team_id + assert again.for_worker_session("w1")[1].actor == "swe-worker" + # budget gate: cap wakes, then refuse until the hour rolls + assert all(again.count_wake(team.team_id, cap=3) for _ in range(3)) + assert again.count_wake(team.team_id, cap=3) is False + + +# -------------------------------------------------------- manager: spawn/digest + +@pytest.fixture +def manager(tmp_path, monkeypatch): + monkeypatch.setenv("COWORKER_STATE_DIR", str(tmp_path / "state")) + ws = tmp_path / "repo" + ws.mkdir() + m = SessionManager(data_dir=tmp_path / "data", workspace=str(ws)) + yield m + + +def test_create_team_fails_closed_on_solo_personas(manager, tmp_path): + from coworker.sessions import SessionRecord + + manager.session_store.save( + SessionRecord( + session_id="lead-sid", + workspace=manager.default_workspace, + model="m", + mode="interactive", + messages=[], + agent="cowork", + ) + ) + result = manager.create_team( + "lead-sid", [{"persona": "cowork"}] + ) # cowork is a solo builtin + assert result["approved"] is False + assert "team-capable" in result["error"] or "team: worker" in result["error"] + assert manager.teams.all() == [] # nothing half-created + + +def test_create_team_prespawns_worker_sessions(manager, monkeypatch): + from coworker.agents.base import Agent + from coworker.sessions import SessionRecord + + worker_agent = Agent( + name="swe-worker", title="SWE", system_prompt="p", team="worker" + ) + monkeypatch.setattr( + "coworker.server.manager.get_agent", lambda name: worker_agent + ) + manager.session_store.save( + SessionRecord( + session_id="lead-sid", + workspace=manager.default_workspace, + model="m", + mode="interactive", + messages=[], + agent="swe-lead", + ) + ) + result = manager.create_team( + "lead-sid", + [{"persona": "swe-worker"}, {"persona": "swe-worker", "model": "other"}], + ) + assert result["approved"] is True + actors = [w["actor"] for w in result["workers"]] + assert actors == ["swe-worker", "swe-worker-2"] # unique actor ids + # pre-spawn = state on disk, zero turns + for w in result["workers"]: + record = manager.session_store.load(w["session_id"]) + assert record is not None + assert record.messages == [] + assert record.team["role"] == "worker" + assert record.team["lead_session"] == "lead-sid" + # the lead session is marked and the registry ties the roster + lead = manager.session_store.load("lead-sid") + assert lead.team["role"] == "lead" + assert manager.teams.for_lead_session("lead-sid") is not None + # second team on the same session refuses + assert manager.create_team("lead-sid", [{"persona": "swe-worker"}])[ + "approved" + ] is False + + +def test_staleness_digest_is_role_scoped(manager, monkeypatch): + from coworker.agents.base import Agent + from coworker.sessions import SessionRecord + from coworker.teams.model import space_for_workspace + + # no team role → no digest (bare wake) + assert manager.team_staleness_digest("nobody") == "" + + worker_agent = Agent(name="swe-worker", title="SWE", system_prompt="p", team="worker") + monkeypatch.setattr("coworker.server.manager.get_agent", lambda name: worker_agent) + manager.session_store.save( + SessionRecord( + session_id="lead-sid", + workspace=manager.default_workspace, + model="m", + mode="interactive", + messages=[], + agent="swe-lead", + ) + ) + manager.create_team("lead-sid", [{"persona": "swe-worker"}]) + space = space_for_workspace(manager.default_workspace) + lead_actor = manager.teams.for_lead_session("lead-sid").lead_actor + item = manager.team_store.create_item( + space, + Actor(id=lead_actor, role=Role.LEAD), + title="Ship it", + criteria="tests green", + ) + digest = manager.team_staleness_digest("lead-sid") + assert "1 open" in digest + assert "no assignee" in digest + _ = item