baphuongna/pi-crew/mailbox-interactive

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)

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.
}

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

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

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;
}

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;
}

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

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
}

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;
}

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;
  }
}

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

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

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