mailbox-interactive · diff
git:20260505.33e7203 to git:20260514.3b16876
299 added, 19 removed. Audit A to A.
---
name: mailbox-interactive
description: Interactive waiting-task and mailbox workflow. Use when implementing or operating respond/nudge/ack/replay/supervisor-contact behavior.
---
# mailbox-interactive
- Use this skill for live coordination between leader and workers.
+ Use this skill for live coordination between leader and workers. Mailbox provides an asynchronous message protocol for steer, follow-up, respond, and nudge operations.
- ## Source patterns distilled
+ ## Mailbox Architecture
- - pi-subagents intercom/contact supervisor: blocking decisions vs non-blocking progress updates
- - pi-crew mailbox: `src/state/mailbox.ts`, `src/extension/team-tool/respond.ts`, `src/extension/team-tool/api.ts`, `src/ui/overlays/mailbox-detail-overlay.ts`, `src/ui/run-action-dispatcher.ts`
- - Waiting state: `src/state/contracts.ts`, `src/runtime/supervisor-contact.ts`, `src/ui/status-colors.ts`
+ ```
+ Worker (waiting) ← mailbox inbox ← Leader (respond)
+ Worker (running) ← mailbox follow-ups ← Leader (followUp)
+ Leader → Worker: steer, followUp, nudge (non-blocking)
+ Worker → Leader: supervisor contact (blocking decision)
+ ```
- ## Rules
+ ### Mailbox file structure
- - Use `waiting` when a task needs leader input and can safely pause.
- - `respond` should write an inbox mailbox message and transition target waiting tasks back to `running`.
- - Mutating mailbox actions must use run locks and re-read state inside the lock.
- - Respect run ownership: foreign sessions cannot respond/resume owned waiting tasks.
- - Mailbox reads should be contained under run state and tolerate missing/empty JSONL files.
- - Acknowledge/read actions are UI/operator state; preserve message history rather than deleting records.
- - Supervisor contact parsed from child stdout should be recorded as events and surfaced in UI without blocking render paths.
+ Each run has a mailbox directory at `.crew/state/runs/<runId>/mailbox/`:
+ - `inbox.jsonl` — incoming messages (to worker)
+ - `outbox.jsonl` — sent messages (from worker)
+ - `steering.jsonl` — steer messages specifically
+ ### Message structure
+
+ ```typescript
+ interface MailboxMessage {
+ id: string;
+ direction: "inbox" | "outbox";
+ from: string; // taskId or "leader"
+ to: string; // taskId or "leader"
+ body: string; // message text
+ status: "pending" | "delivered" | "acknowledged" | "rejected";
+ priority: "low" | "normal" | "high";
+ sentAt: string; // ISO timestamp
+ deliveredAt?: string;
+ data?: Record<string, unknown>; // source, correlation, etc.
+ }
+ ```
+
+ ## Core Operations
+
+ ### 1. respond — Leader responds to waiting worker
+
+ ```typescript
+ // Respond writes to inbox and transitions task back to running
+ async function respond(runId: string, taskId: string, body: string, priority = "normal") {
+ // 1. Write inbox message
+ const message = appendInboxMessage(manifest, { taskId, body, priority });
+
+ // 2. Re-read state inside lock
+ const { tasks } = loadRunManifestById(cwd, runId);
+ const task = tasks.find(t => t.id === taskId);
+
+ // 3. Verify task is waiting
+ if (task.status !== "waiting") {
+ throw new Error(`Cannot respond to non-waiting task: ${task.status}`);
+ }
+
+ // 4. Transition task back to running
+ const updated = { ...task, status: "running", waitingSince: undefined };
+ saveRunTasks(manifest, [updated]);
+
+ // 5. Emit event
+ appendEvent(eventsPath, { type: "task.responded", taskId, message: body });
+
+ return message;
+ }
+ ```
+
+ ### 2. steer — Live agent steering (non-blocking)
+
+ ```typescript
+ // Steer sends a message to a running live agent
+ async function steerLiveAgent(agentId: string, message: string) {
+ const handle = getLiveAgent(agentId);
+ if (!handle) throw new Error(`Live agent '${agentId}' not found`);
+
+ // If session.steer is available, deliver immediately
+ if (typeof handle.session.steer === "function") {
+ await handle.session.steer(message);
+ handle.updatedAt = new Date().toISOString();
+ return handle;
+ }
+
+ // Otherwise, queue for delivery when session becomes ready
+ handle.pendingSteers.push(message);
+ return handle;
+ }
+ ```
+
+ ### 3. followUp — Non-blocking follow-up to running agent
+
+ ```typescript
+ async function followUpLiveAgent(agentId: string, prompt: string) {
+ const handle = getLiveAgent(agentId);
+ if (!handle) throw new Error(`Live agent '${agentId}' not found`);
+
+ if (typeof handle.session.prompt === "function") {
+ await handle.session.prompt(prompt, { source: "api", expandPromptTemplates: false });
+ handle.updatedAt = new Date().toISOString();
+ return handle;
+ }
+
+ handle.pendingFollowUps.push(prompt);
+ return handle;
+ }
+ ```
+
+ ### 4. nudge — Ask agent to report status
+
+ ```typescript
+ function nudgeAgent(manifest: TeamRunManifest, agentId: string, message?: string) {
+ const agent = readCrewAgents(manifest).find(a => a.id === agentId || a.taskId === agentId);
+ if (!agent) throw new Error(`Agent '${agentId}' not found`);
+
+ const text = message ?? "Please report your current status, blocker, or smallest next step.";
+
+ // Write to mailbox
+ const mailboxMessage = appendSteeringMessage(manifest, {
+ taskId: agent.taskId,
+ body: text,
+ priority: "normal",
+ data: { source: "nudge-agent" }
+ });
+
+ // Emit event
+ appendEvent(manifest.eventsPath, {
+ type: "agent.nudged",
+ runId: manifest.runId,
+ taskId: agent.taskId,
+ message: text,
+ data: { agentId: agent.id, mailboxMessageId: mailboxMessage.id }
+ });
+
+ return mailboxMessage;
+ }
+ ```
+
+ ## Supervisor Contact
+
+ Workers can contact the leader for blocking decisions:
+
+ ```typescript
+ // Worker stdout contains supervisor contact pattern:
+ // @supervisor need_confirmation: <decision-type>
+ // @supervisor need_input: <input-prompt>
+
+ // Parsed by parseSupervisorContactFromLine() in supervisor-contact.ts
+ function parseSupervisorContact(line: string): SupervisorContact | null {
+ const match = line.match(/@supervisor\s+(\w+):\s*(.*)/);
+ if (!match) return null;
+ return { type: match[1], prompt: match[2] };
+ }
+
+ // Recorded as events and surfaced in UI
+ appendEvent(eventsPath, {
+ type: "supervisor.contact",
+ runId: manifest.runId,
+ taskId: task.id,
+ message: contact.prompt,
+ data: { contactType: contact.type }
+ });
+ ```
+
+ **Contact types:**
+ - `need_confirmation` — worker needs explicit approval
+ - `need_input` — worker needs a text response
+ - `need_selection` — worker needs to choose from options
+
+ ## Queue Depth and Backpressure
+
+ Mailbox queues can grow large with pending messages:
+
+ ```typescript
+ const MAX_PENDING_STEERS = 50;
+ const MAX_PENDING_FOLLOWUPS = 50;
+
+ // When queue exceeds limit, oldest messages are dropped
+ if (handle.pendingSteers.length >= MAX_PENDING_STEERS) {
+ handle.pendingSteers.shift(); // drop oldest
+ }
+ ```
+
+ **Backpressure signals:**
+ - If `pendingSteers.length > 20`, warn in UI
+ - If `pendingFollowUps.length > 20`, warn in UI
+
+ ## Corrupt JSONL Handling
+
+ Mailbox JSONL files can become corrupt on crash. Handle gracefully:
+
+ ```typescript
+ function readMailboxMessages(path: string): MailboxMessage[] {
+ if (!fs.existsSync(path)) return [];
+
+ const lines = fs.readFileSync(path, "utf-8").split("\n");
+ const messages: MailboxMessage[] = [];
+
+ for (const line of lines) {
+ if (!line.trim()) continue;
+ try {
+ messages.push(JSON.parse(line) as MailboxMessage);
+ } catch {
+ // Skip corrupt lines: partial write from crash
+ // Try to recover by reading up to the last valid JSON object
+ continue;
+ }
+ }
+
+ return messages;
+ }
+ ```
+
+ ## Timeout Patterns
+
+ ### Responding with timeout
+
+ ```typescript
+ async function respondWithTimeout(
+ runId: string,
+ taskId: string,
+ body: string,
+ timeoutMs = 30_000
+ ): Promise<MailboxMessage> {
+ const controller = new AbortController();
+ const timeout = setTimeout(() => controller.abort(), timeoutMs);
+
+ try {
+ const message = await respond(runId, taskId, body);
+ clearTimeout(timeout);
+ return message;
+ } catch (error) {
+ clearTimeout(timeout);
+ if (error.name === "AbortError") {
+ throw new Error(`Respond timed out after ${timeoutMs}ms`);
+ }
+ throw error;
+ }
+ }
+ ```
+
+ ### Non-blocking follow-up (no wait)
+
+ ```typescript
+ // Fire-and-forget: send message and return immediately
+ function followUpAsync(agentId: string, prompt: string): void {
+ const handle = getLiveAgent(agentId);
+ if (!handle) return;
+
+ if (typeof handle.session.prompt === "function") {
+ // Non-awaited: delivery happens in background
+ void handle.session.prompt(prompt, { source: "api", expandPromptTemplates: false })
+ .catch(() => { /* log if needed */ });
+ } else {
+ handle.pendingFollowUps.push(prompt);
+ }
+ }
+ ```
+
+ ## Run Ownership Enforcement
+
+ Foreign sessions cannot mutate a run's mailbox:
+
+ ```typescript
+ // In respond.ts / cancel.ts
+ function verifyRunOwnership(manifest: TeamRunManifest, sessionId: string, force = false) {
+ if (!force && manifest.ownerSessionId && manifest.ownerSessionId !== sessionId) {
+ throw new Error(
+ `Run ${manifest.runId} is owned by session ${manifest.ownerSessionId}. ` +
+ `Cannot mutate from session ${sessionId}. Use force=true to override.`
+ );
+ }
+ }
+ ```
+
## Anti-patterns
- - Resuming non-waiting tasks via `respond`.
- - Injecting mailbox messages into a foreign owned run.
- - Treating every progress update as a blocking supervisor decision.
- - Reading large mailbox files synchronously in hot render paths.
+ - **Resuming non-waiting tasks**: `respond` only works on `waiting` tasks. Resuming `running` tasks corrupts state.
+ - **Injecting mailbox messages into foreign runs**: Always verify `ownerSessionId` before mutating.
+ - **Treating every progress update as blocking**: Use `followUp` (non-blocking) instead of `respond` for status updates.
+ - **Reading large mailbox files in hot paths**: Cache mailbox counts; don't read JSONL on every render tick.
+ - **Not handling corrupt JSONL**: Skip malformed lines; don't fail the whole read.
+ - **Losing pending messages on session switch**: Pending steers/followups are stored in-memory in the handle. They survive session fork but not session death.
+ ---
+
+ ## Source patterns
+
+ - `src/state/mailbox.ts` — appendInboxMessage, appendSteeringMessage, readMailboxMessages
+ - `src/extension/team-tool/respond.ts` — respond tool handler, ownership verification
+ - `src/extension/team-tool/cancel.ts` — cancel with mailbox cleanup
+ - `src/extension/team-tool/api.ts` — steer-agent, follow-up-agent, nudge-agent
+ - `src/runtime/live-agent-manager.ts` — pendingSteers, pendingFollowUps, steerLiveAgent, followUpLiveAgent
+ - `src/runtime/supervisor-contact.ts` — parseSupervisorContactFromLine, recordSupervisorContact
+ - `src/ui/overlays/mailbox-detail-overlay.ts` — mailbox UI
+
+ ---
+
## Verification
```bash
cd pi-crew
+
+ # Verify mailbox structure
+ ls .crew/state/runs/<runId>/mailbox/ 2>/dev/null || echo "No mailbox yet"
+
+ # Test respond tool
+ node --experimental-strip-types --test test/unit/respond-tool.test.ts
+
+ # Test mailbox overlay
+ node --experimental-strip-types --test test/unit/mailbox-detail-overlay.test.ts
+
+ # TypeScript
npx tsc --noEmit
- node --experimental-strip-types --test test/unit/respond-tool.test.ts test/unit/mailbox-detail-overlay.test.ts test/unit/mailbox-compose-overlay.test.ts test/unit/supervisor-contact.test.ts
+
+ # All tests
npm test
```
-