mailbox-interactive

name	mailbox-interactive
description	Interactive waiting-task and mailbox workflow.
origin	pi-crew
triggers	["respond to worker","nudge agent","mailbox message","supervisor contact","waiting task"]

mailbox-interactive

Use this skill for live coordination between leader and workers. Mailbox provides an asynchronous message protocol for steer, follow-up, respond, and nudge operations.

Mailbox Architecture

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)

Mailbox file structure

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

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

// 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)

// 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

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

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:

// 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:

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:

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

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)

// 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:

// 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.`
    );
  }
}

Enforcement — Mailbox Interactive Gate

Before responding to or mutating mailbox state, verify:

• Target task status is "waiting" (respond only works on waiting tasks)
• ownerSessionId matches current session (ownership verified)
• Run status is not terminal (do not respond to completed/failed/cancelled)
• Corrupt JSONL handled gracefully (skip malformed lines)
• Backpressure respected (queue depth below MAX_PENDING limits)

If ANY answer is NO → Stop. Verify mailbox state before mutating.

Anti-patterns

• 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

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

# All tests
npm test