sgtbeatdown-pi-tasks

内容来源：README.md（说明文档） · 原始地址 · 查看安装指南

原始内容

# pi-tasks

Structured task communication for Pi agents.

`pi-tasks` lets one Pi session hand work to another agent, session, process, or
worker and get durable structured results back. The core model is transport- and
runner-neutral: local sessions, HTTP workers, queues, hosted control planes, and
terminal multiplexers can all implement the same task lifecycle.

## what this is

A task lifecycle and communication layer:

1. create a structured assignment
2. persist task state in a store
3. dispatch the assignment through a transport
4. let the sender keep working
5. record status/results in structured files/data, not terminal prose
6. require the assignee to finish through the task protocol

## what this is not

This is **not** a subagent spawner. Use any process manager, terminal/session
manager, HTTP worker pool, queue, cron job, or custom runner to create/host
agents. `pi-tasks` only gives those agents a shared task protocol.

## mental model

```text
parent pi session
  agent_task_send
    ↓
  TaskStore      creates task + durable refs
  TaskTransport  delivers assignment text to target
    ↓
target runner/session/worker
  receives pi.task.assignment.v1
  does work
  completes task through agent_task_done or equivalent store/server completion API
    ↓
parent pi session
  agent_task_status / agent_task_wait / agent_task_inbox
```

The `to` field is transport-specific:

- session name
- worker id
- queue/topic name
- tmux pane
- process id
- service-specific address

## requirements for any runner

For task communication to work, you need exactly these pieces:

1. **running target** — an already-open Pi session, worker, process, or service.
   `pi-tasks` does not open or close it.
2. **transport** — code that delivers the assignment text to that target. This
   can be `tmux paste-buffer`, HTTP POST, queue publish, file write, ssh, etc.
3. **reachable task store** — parent and target must read/write the same task
   state. With `createFilesystemTaskStore`, that means the same cwd and shared
   `.pi/tasks` directory. For multi-host setups without shared disk, implement a
   remote `TaskStore`.
4. **completion capability** — Pi targets need this extension loaded so they have
   `agent_task_done`. Non-Pi workers need an equivalent completion API/store
   write.
5. **preserved task id** — the target must complete the exact `taskId` from the
   `pi.task.assignment.v1` assignment.

If those are true, session lifecycle is irrelevant. Open agents with tmux,
screen, ssh, a process manager, or manually; `pi-tasks` only handles the task
protocol once the agents exist.

## install

Install the current local checkout during development:

```bash
pi install /Users/home/Dev/wolfpack-pi-tasks
```

Install from npm when using a published release:

```bash
pi install npm:@sgtbeatdown/pi-tasks
```

Temporary one-off run from a local checkout or npm package:

```bash
pi -e /Users/home/Dev/wolfpack-pi-tasks
pi -e npm:@sgtbeatdown/pi-tasks
```

Every participating Pi session must load this extension. Before sending, verify
the target has `agent_task_done` and can reach the same task store. If not, fail
fast and instruct setup before dispatch; do not create a task that can only time
out.

## quick start: local/custom transport

Most integrations reuse the filesystem store and provide only a transport:

```ts
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
import { registerAgentTaskTools } from "@sgtbeatdown/pi-tasks/src/extension";
import { createFilesystemTaskStore } from "@sgtbeatdown/pi-tasks/src/stores/filesystem";
import type { TaskTransport } from "@sgtbeatdown/pi-tasks/src/task-communication";

function createCliTransport(pi: ExtensionAPI): TaskTransport {
  return {
    name: "local-cli",
    getCurrentSessionName(env) {
      return env.PI_TASK_SESSION ?? env.USER ?? "unknown-session";
    },
    async dispatchTask({ target, assignment, signal }) {
      const result = await pi.exec("my-session-send", [target, assignment], { signal });
      if (result.code === 0) return { ok: true };
      return {
        ok: false,
        message: result.stderr || result.stdout || "my-session-send failed",
        retryable: true,
      };
    },
  };
}

export default function extension(pi: ExtensionAPI): void {
  registerAgentTaskTools(pi, {
    store: createFilesystemTaskStore({ tasksDir: ".pi/tasks" }),
    transport: createCliTransport(pi),
  });
}
```

Then call `agent_task_send` from the parent session:

```json
{
  "to": "worker-a",
  "task": "inspect the auth middleware and report risks",
  "metadata": {
    "phaseId": "phase-1",
    "issueId": "auth-boundary",
    "role": "reviewer",
    "verificationTier": "focused"
  },
  "contextRefs": [
    { "path": ".plans/current.md", "selector": "L10-L40", "required": true, "purpose": "scope" }
  ],
  "preflight": { "requireReachable": false },
  "timeoutMs": 1800000,
  "onCompletePrompt": "review the worker's findings before reporting back to the user"
}
```

The parent should keep working. normal delegation requests are fire-and-forget.
Do not call `agent_task_wait` after dispatch; use it only when the current user
message explicitly asks to block for the result now. When the task reaches a
terminal state, the extension updates the inbox status and, at an
idle/no-pending-message boundary, injects a reminder to call
`agent_task_inbox({ ack: true })`. If `onCompletePrompt` was set, that
sender-defined parent-side reminder is included in the idle notification and
compact inbox/status result. Do **not** infer completion from terminal output.

## tmux recipe: already-open pi agents

Use this when you manually open Pi agents in tmux panes and only need task
communication between them.

### 1. create a tmux transport extension

Create `.pi/extensions/pi-tasks-tmux.ts` in the project:

```ts
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
import { registerAgentTaskTools } from "@sgtbeatdown/pi-tasks/src/extension";
import { createFilesystemTaskStore } from "@sgtbeatdown/pi-tasks/src/stores/filesystem";
import type { TaskTransport } from "@sgtbeatdown/pi-tasks/src/task-communication";

function createTmuxTransport(pi: ExtensionAPI): TaskTransport {
  return {
    name: "tmux",
    getCurrentSessionName(env) {
      return env.PI_TASK_SESSION ?? env.TMUX_PANE ?? `pid-${process.pid}`;
    },
    async dispatchTask({ target, assignment, task, signal }) {
      const bufferName = `pi-task-${task.id}`;
      const setBuffer = await pi.exec("tmux", ["set-buffer", "-b", bufferName, assignment], { signal });
      if (setBuffer.code !== 0) {
        return {
          ok: false,
          message: setBuffer.stderr || setBuffer.stdout || "tmux set-buffer failed",
          retryable: true,
        };
      }

      const paste = await pi.exec("tmux", ["paste-buffer", "-d", "-b", bufferName, "-t", target], { signal });
      if (paste.code !== 0) {
        return {
          ok: false,
          message: paste.stderr || paste.stdout || "tmux paste-buffer failed",
          retryable: true,
        };
      }

      const enter = await pi.exec("tmux", ["send-keys", "-t", target, "Enter"], { signal });
      if (enter.code !== 0) {
        return {
          ok: false,
          message: enter.stderr || enter.stdout || "tmux send-keys failed",
          retryable: true,
        };
      }

      return { ok: true };
    },
  };
}

export default function extension(pi: ExtensionAPI): void {
  registerAgentTaskTools(pi, {
    // All tmux panes must run in the same project cwd so they share .pi/tasks.
    store: createFilesystemTaskStore({ tasksDir: ".pi/tasks" }),
    transport: createTmuxTransport(pi),
  });
}
```

This transport does not open panes. It only pastes assignment text into an
already-running pane and presses Enter.

### 2. open the agents yourself

Start each Pi agent in the same repo/cwd and load the extension. Give each one a
stable session name:

```bash
# pane 1
PI_TASK_SESSION=parent pi -e ./.pi/extensions/pi-tasks-tmux.ts

# pane 2
PI_TASK_SESSION=worker-a pi -e ./.pi/extensions/pi-tasks-tmux.ts
```

If the extension is installed with `pi install`, you can omit `-e`; the important
parts are same cwd, same `.pi/tasks`, and distinct `PI_TASK_SESSION` values.

### 3. find the target pane id

From inside tmux:

```bash
tmux list-panes -a -F '#{pane_id} #{pane_current_path} #{pane_title}'
```

Pane ids look like `%12`. Use that pane id as `agent_task_send.to`.

### 4. send a task

From the parent Pi session, call:

```json
{
  "to": "%12",
  "task": "inspect the auth middleware and report risks",
  "metadata": { "phaseId": "phase-1", "issueId": "auth-review", "role": "reviewer", "verificationTier": "focused" },
  "contextRefs": [{ "path": ".plans/current.md", "required": false, "purpose": "scope" }],
  "timeoutMs": 1800000
}
```

Expected flow:

1. parent creates `.pi/tasks/<taskId>/...`
2. tmux transport pastes the assignment into pane `%12`
3. worker receives `pi.task.assignment.v1`
4. worker does the work
5. worker calls `agent_task_done` with that `taskId`
6. parent reads the result later via `agent_task_status` or `agent_task_inbox`;
   use `agent_task_wait` only when the current user message explicitly asks to
   block

### tmux troubleshooting

- If the target cannot complete the task, confirm it loaded the extension and has
  `agent_task_done` available.
- If status never changes, confirm parent and worker are in the same cwd and both
  can see the same `.pi/tasks/<taskId>` directory.
- If dispatch fails, run `tmux paste-buffer` manually against the same target id;
  the pane id may be stale.
- Do not use terminal output as the source of truth. The task result is the store
  state, not visible prose in the pane.

## using with non-pi workers

A non-Pi worker can participate if it implements the same lifecycle:

1. receive `pi.task.assignment.v1`
2. do the assigned work
3. write/report a terminal result with the same status and payload semantics
4. never overwrite an already-terminal task

Example HTTP transport from a Pi extension:

```ts
import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
import { registerAgentTaskTools } from "@sgtbeatdown/pi-tasks/src/extension";
import { createFilesystemTaskStore } from "@sgtbeatdown/pi-tasks/src/stores/filesystem";
import type { TaskTransport } from "@sgtbeatdown/pi-tasks/src/task-communication";

const httpTransport: TaskTransport = {
  name: "http",
  getCurrentSessionName(env) {
    return env.PI_TASK_SESSION ?? "unknown-session";
  },
  async dispatchTask({ target, assignment, task, signal }) {
    const response = await fetch(`https://tasks.example/sessions/${target}/assignments`, {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: JSON.stringify({ taskId: task.id, assignment }),
      signal,
    });

    if (response.ok) return { ok: true };

    return {
      ok: false,
      message: await response.text(),
      retryable: response.status >= 500,
    };
  },
};

export default function extension(pi: ExtensionAPI): void {
  registerAgentTaskTools(pi, {
    store: createFilesystemTaskStore({ tasksDir: ".pi/tasks" }),
    transport: httpTransport,
  });
}
```

For non-Pi workers, expose a completion endpoint or let workers use your store
implementation directly. The worker should report a terminal result equivalent
to:

```json
{
  "taskId": "task_...",
  "status": "completed",
  "summary": "finished the assigned work",
  "result": {}
}
```

## file inbox / polling workers

A pushed message is not required. A transport can durably drop assignments into a
file inbox or queue and return success once the assignment is written.

One simple design:

- store tasks in `.pi/tasks`
- transport writes assignment text to `.pi/task-inbox/<target>/<taskId>.txt`
- worker watches its inbox directory
- worker reads the assignment, does the work, then completes through a store API
  or by invoking a Pi session that has `agent_task_done`

Completion is separate from dispatch. Dispatch means “assignment delivered,” not
“work finished.”

## when to implement a custom store

Do **not** implement a custom store just because you have a new dispatch
mechanism. Implement `TaskStore` only when task state itself must be somewhere
else:

- central HTTP service
- Redis/Postgres/SQLite shared by multiple machines
- multi-host queue system
- hosted control plane

If only delivery changes, keep the filesystem store.

## included wolfpack transport

This package includes a Wolfpack transport for convenience. It is only a delivery
adapter:

```bash
wolfpack session send <target-session> <assignment>
```

It does not own storage. The default exported extension uses the generic
filesystem store at `.pi/tasks/` plus the included Wolfpack transport. Preflight
uses structured Wolfpack status only:

```bash
wolfpack session status <target-session> --json
```

Only structured status with `terminal.exists === true`, `terminal.alive === true`,
and `terminal.status === "ready"` passes `transport_reachable`. Structured
`SESSION_NOT_FOUND`, `AMBIGUOUS_SELECTOR`, and `SESSION_DEAD` errors fail
reachability. `BACKEND_UNAVAILABLE`, invalid JSON, and command errors are
unavailable unless `preflight.requireReachable` is true. If
`preflight.requiredProjectDir` is supplied, the returned target `projectDir` or
`projectPath` is compared with resolved paths; missing target project facts fail
that required check. Terminal output/prose is never used as the source of truth.

If you are using Wolfpack, install this extension in every participating Pi
session and target a session name/id with `agent_task_send`. Do not use
cross-repo `agent_task_send` with the filesystem store because separate project
roots do not share task state. Until a shared/global store exists, direct
`wolfpack session send` is only a fallback instruction channel, not a task
completion protocol. Do not use symlinks to imitate shared task storage.

If you are not using Wolfpack, register the tools with your own `{ store,
transport }` composition.

## tools

| tool | purpose |
| --- | --- |
| `agent_task_send` | preflight, create, and dispatch a task through the configured transport; optionally set workflow `metadata`, `contextRefs`, `preflight`, and `onCompletePrompt` |
| `agent_task_status` | read compact structured status for one task, including any sender-defined `onCompletePrompt` |
| `agent_task_wait` | blocking wait for a terminal result; do not call after dispatch unless the current user message explicitly asks to block |
| `agent_task_inbox` | list terminal tasks for the current parent session, including any sender-defined `onCompletePrompt` |
| `agent_task_cancel` | cancel a non-terminal task |
| `agent_task_done` | assignee-side structured completion; terminates the target response |

### completion contract

Assigned Pi agents must call `agent_task_done` as their final action:

```json
{
  "taskId": "task_...",
  "status": "completed",
  "summary": "inspected auth middleware; no bypass found",
  "result": {
    "issueId": "auth-boundary",
    "verdict": "completed",
    "changedFiles": [],
    "verification": [
      { "command": "bun test tests/auth.test.ts", "status": "passed", "exitCode": 0, "summary": "focused auth tests" }
    ],
    "next": "ready for parent review"
  }
}
```

Keep `summary` at or below 1200 characters. Put compact machine-readable details
under `result`, using `issueId`, `verdict`, `changedFiles`, `verification`,
`blockers`, `risks`, and `next` when useful. Verification entries should include
the exact command, status, exit code, and short summary where applicable.

Non-Pi workers should complete through an equivalent store/server API with the
same terminal status and payload semantics.

Valid terminal statuses:

- `completed`
- `failed`
- `cancelled`
- `timed_out`
- `rejected`

Terminal completion is first-writer-wins.

### preflight and workflow fields

`agent_task_send` runs protocol/store preflight before assignment delivery:

- empty target strings fail immediately
- `metadata.issueId` blocks duplicate active tasks for the same target + issue
- required `contextRefs` must be project-local and readable
- `requiredModel` and `requireIdle` currently report `unavailable` because Pi does not expose target facts yet
- transports may implement `preflightTarget`; otherwise liveness is `unavailable`, or `failed` when `preflight.requireReachable` is true
- `preflight.requiredProjectDir` checks target project facts returned by transport preflight, not the parent cwd

Failed preflight without `idempotencyKey` returns an ephemeral rejected tool
result: no task directory is created and assignment text is not delivered. Failed
preflight with `idempotencyKey` creates or reuses a durable terminal `rejected`
task with `error.code = "preflight_failed"`, saved `preflight` checks, no
`assignmentRef`, and no delivered assignment text.

`contextRefs.selector` is stored and sent as opaque metadata in v1. The sender
validates only path containment/readability, not line ranges, headings, or JSON
pointers.

## task protocol

Assignments are delivered as human-readable text plus a structured envelope with:

- `type: "pi.task.assignment.v1"`
- `taskId`
- `fromSession`
- `instructions`
- optional workflow `metadata`
- optional bounded `contextRefs` that the assignee can read explicitly
- protocol requirements telling the assignee to use `agent_task_done`

Task state includes:

- `pending`
- `dispatched`
- `running`
- terminal status
- assignment/result refs
- event history
- optional preflight checks/result
- optional workflow metadata and context refs
- optional sender-defined `onCompletePrompt` for parent follow-up
- optional result payload/artifacts/error

Filesystem layout for dispatched tasks:

```text
<tasksDir>/<taskId>/
├── task.json
├── assignment.json
├── events.jsonl
└── result.json
```

Durable idempotent preflight-rejected tasks omit `assignment.json` and have
`assignmentRef: undefined` because no assignment was delivered. Non-idempotent
preflight failures are ephemeral and create no task directory.

Generic filesystem storage defaults to `.pi/tasks/` when used directly.

## postmortem metrics

Analyze an existing filesystem task-store root without reading terminal output:

```bash
bun run task-metrics /path/to/project/.pi/tasks
```

The command prints a deterministic JSON report. The production API is also
available directly:

```ts
import { analyzeTaskStoreMetrics } from "@sgtbeatdown/pi-tasks/src/metrics";

const report = await analyzeTaskStoreMetrics("/path/to/project/.pi/tasks");
```

The report includes status counts; durable preflight rejections; failed and
unavailable preflight checks by check name; grouping counts for `phaseId`,
`issueId`, and `rootCause`; and `loopsPerIssueId` as the raw task count for each
issue. Character totals measure `taskText`, parsed `assignment.json` values,
object-form `instructions`, result summaries, and compact `JSON.stringify` sizes
for the value under `result.result`. A largest-first prompt/result list is capped
at 10 entries; each result size is its summary plus serialized structured payload.

Verification counts come **only** from structured
`result.result.verification[]` entries. The report does not infer verification
truth from task/result prose. Missing or malformed artifacts produce structured
`diagnostics` and do not stop analysis of other valid task directories.

## task board

Print a compact issue/root-cause board from the same structured artifacts:

```bash
bun run task-metrics /path/to/project/.pi/tasks --board
```

Or call the production API:

```ts
import { buildTaskBoard } from "@sgtbeatdown/pi-tasks/src/task-board";

const board = await buildTaskBoard("/path/to/project/.pi/tasks");
```

The JSON groups complete `phaseId`/`issueId`/`rootCause` tuples, puts groups
with active tasks first, then sorts by latest update/completion and identifiers.
It includes task/status counts, structured verification/blocker/risk counts,
and repeated-issue/shared-root-cause indicators. Output defaults to 10 groups
and 20 task ids per group.

Legacy tasks with missing grouping metadata are never hidden or assigned inferred
metadata. `totalTaskRecords`, `groupedTaskCount`, and `ungroupedTaskCount`
reconcile coverage; `ungrouped` contains status counts, missing-field reason
counts, and up to 20 task ids. Missing/malformed artifacts remain in structured
`diagnostics`.

## architecture

`pi-tasks` separates storage from delivery:

```ts
interface TaskStore {
  createOrReuseDispatchedTask(...): Promise<...>;
  createOrReusePreflightRejectedTask(...): Promise<...>;
  readTask(...): Promise<...>;
  readTaskResult(...): Promise<...>;
  expireTaskIfOverdue(...): Promise<...>;
  waitForTask(...): Promise<...>;
  listInbox(...): Promise<...>;
  findActiveTaskByIssueId(...): Promise<...>;
  ackTask(...): Promise<...>;
  cancelTask(...): Promise<...>;
  completeTask(...): Promise<...>;
}

interface TaskTransport {
  getCurrentSessionName(env): string;
  preflightTarget?(input): Promise<TaskPreflightResult>;
  dispatchTask(input): Promise<DispatchTaskResult>;
}
```

Registration composes both:

```ts
registerAgentTaskTools(pi, {
  store: createFilesystemTaskStore({ tasksDir: ".pi/tasks" }),
  transport: myTransport,
});
```

## exported building blocks

- `src/extension.ts`
  - `registerAgentTaskTools`
  - `createDefaultTaskCommunicationLayer`
- `src/task-communication.ts`
  - `TaskStore`
  - `TaskTransport`
  - `TaskCommunicationLayer`
  - `PreflightTargetInput`
- `src/stores/filesystem.ts`
  - `createFilesystemTaskStore`
- `src/transports/wolfpack.ts`
  - `createWolfpackTaskTransport`
- `src/protocol.ts`
  - `buildAssignment`
  - `compactTaskResult`
  - `validateStructuredTaskResult`
- `src/preflight.ts`
  - `runTaskPreflight`
- `src/metrics.ts`
  - `analyzeTaskStoreMetrics`
- `src/task-board.ts`
  - `buildTaskBoard`
- `src/task-artifacts.ts`
  - `readTaskStoreArtifacts`
- `src/store.ts`
  - low-level filesystem store functions, if you need finer control than
    `createFilesystemTaskStore`

## development

```bash
bun install
bun test
bun run typecheck
```

## current limitations

- no built-in HTTP/Redis transport yet
- no runtime transport selector yet; compose a store/transport in an extension
- default extension uses the generic filesystem store plus the included Wolfpack transport
- Wolfpack preflight depends on the structured `wolfpack session status <target> --json` contract
- the task board is a bounded read-only summary, not a shared registry or project-management system
- local development install is `pi install /Users/home/Dev/wolfpack-pi-tasks`; published releases can use `pi install npm:@sgtbeatdown/pi-tasks`

pi-tasks

Structured task communication for Pi agents.

pi-tasks lets one Pi session hand work to another agent, session, process, or worker and get durable structured results back. The core model is transport- and runner-neutral: local sessions, HTTP workers, queues, hosted control planes, and terminal multiplexers can all implement the same task lifecycle.

what this is

A task lifecycle and communication layer:

create a structured assignment
persist task state in a store
dispatch the assignment through a transport
let the sender keep working
record status/results in structured files/data, not terminal prose
require the assignee to finish through the task protocol

what this is not

This is not a subagent spawner. Use any process manager, terminal/session manager, HTTP worker pool, queue, cron job, or custom runner to create/host agents. pi-tasks only gives those agents a shared task protocol.

mental model

parent pi session
  agent_task_send
    ↓
  TaskStore      creates task + durable refs
  TaskTransport  delivers assignment text to target
    ↓
target runner/session/worker
  receives pi.task.assignment.v1
  does work
  completes task through agent_task_done or equivalent store/server completion API
    ↓
parent pi session
  agent_task_status / agent_task_wait / agent_task_inbox

The to field is transport-specific:

session name
worker id
queue/topic name
tmux pane
process id
service-specific address

requirements for any runner

For task communication to work, you need exactly these pieces:

running target — an already-open Pi session, worker, process, or service. pi-tasks does not open or close it.
transport — code that delivers the assignment text to that target. This can be tmux paste-buffer, HTTP POST, queue publish, file write, ssh, etc.
reachable task store — parent and target must read/write the same task state. With createFilesystemTaskStore, that means the same cwd and shared .pi/tasks directory. For multi-host setups without shared disk, implement a remote TaskStore.
completion capability — Pi targets need this extension loaded so they have agent_task_done. Non-Pi workers need an equivalent completion API/store write.
preserved task id — the target must complete the exact taskId from the pi.task.assignment.v1 assignment.

If those are true, session lifecycle is irrelevant. Open agents with tmux, screen, ssh, a process manager, or manually; pi-tasks only handles the task protocol once the agents exist.

install

Install the current local checkout during development:

pi install /Users/home/Dev/wolfpack-pi-tasks

Install from npm when using a published release:

pi install npm:@sgtbeatdown/pi-tasks

Temporary one-off run from a local checkout or npm package:

pi -e /Users/home/Dev/wolfpack-pi-tasks
pi -e npm:@sgtbeatdown/pi-tasks

Every participating Pi session must load this extension. Before sending, verify the target has agent_task_done and can reach the same task store. If not, fail fast and instruct setup before dispatch; do not create a task that can only time out.

quick start: local/custom transport

Most integrations reuse the filesystem store and provide only a transport:

import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
import { registerAgentTaskTools } from "@sgtbeatdown/pi-tasks/src/extension";
import { createFilesystemTaskStore } from "@sgtbeatdown/pi-tasks/src/stores/filesystem";
import type { TaskTransport } from "@sgtbeatdown/pi-tasks/src/task-communication";

function createCliTransport(pi: ExtensionAPI): TaskTransport {
  return {
    name: "local-cli",
    getCurrentSessionName(env) {
      return env.PI_TASK_SESSION ?? env.USER ?? "unknown-session";
    },
    async dispatchTask({ target, assignment, signal }) {
      const result = await pi.exec("my-session-send", [target, assignment], { signal });
      if (result.code === 0) return { ok: true };
      return {
        ok: false,
        message: result.stderr || result.stdout || "my-session-send failed",
        retryable: true,
      };
    },
  };
}

export default function extension(pi: ExtensionAPI): void {
  registerAgentTaskTools(pi, {
    store: createFilesystemTaskStore({ tasksDir: ".pi/tasks" }),
    transport: createCliTransport(pi),
  });
}

Then call agent_task_send from the parent session:

{
  "to": "worker-a",
  "task": "inspect the auth middleware and report risks",
  "metadata": {
    "phaseId": "phase-1",
    "issueId": "auth-boundary",
    "role": "reviewer",
    "verificationTier": "focused"
  },
  "contextRefs": [
    { "path": ".plans/current.md", "selector": "L10-L40", "required": true, "purpose": "scope" }
  ],
  "preflight": { "requireReachable": false },
  "timeoutMs": 1800000,
  "onCompletePrompt": "review the worker's findings before reporting back to the user"
}

The parent should keep working. normal delegation requests are fire-and-forget. Do not call agent_task_wait after dispatch; use it only when the current user message explicitly asks to block for the result now. When the task reaches a terminal state, the extension updates the inbox status and, at an idle/no-pending-message boundary, injects a reminder to call agent_task_inbox({ ack: true }). If onCompletePrompt was set, that sender-defined parent-side reminder is included in the idle notification and compact inbox/status result. Do not infer completion from terminal output.

tmux recipe: already-open pi agents

Use this when you manually open Pi agents in tmux panes and only need task communication between them.

1. create a tmux transport extension

Create .pi/extensions/pi-tasks-tmux.ts in the project:

import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
import { registerAgentTaskTools } from "@sgtbeatdown/pi-tasks/src/extension";
import { createFilesystemTaskStore } from "@sgtbeatdown/pi-tasks/src/stores/filesystem";
import type { TaskTransport } from "@sgtbeatdown/pi-tasks/src/task-communication";

function createTmuxTransport(pi: ExtensionAPI): TaskTransport {
  return {
    name: "tmux",
    getCurrentSessionName(env) {
      return env.PI_TASK_SESSION ?? env.TMUX_PANE ?? `pid-${process.pid}`;
    },
    async dispatchTask({ target, assignment, task, signal }) {
      const bufferName = `pi-task-${task.id}`;
      const setBuffer = await pi.exec("tmux", ["set-buffer", "-b", bufferName, assignment], { signal });
      if (setBuffer.code !== 0) {
        return {
          ok: false,
          message: setBuffer.stderr || setBuffer.stdout || "tmux set-buffer failed",
          retryable: true,
        };
      }

      const paste = await pi.exec("tmux", ["paste-buffer", "-d", "-b", bufferName, "-t", target], { signal });
      if (paste.code !== 0) {
        return {
          ok: false,
          message: paste.stderr || paste.stdout || "tmux paste-buffer failed",
          retryable: true,
        };
      }

      const enter = await pi.exec("tmux", ["send-keys", "-t", target, "Enter"], { signal });
      if (enter.code !== 0) {
        return {
          ok: false,
          message: enter.stderr || enter.stdout || "tmux send-keys failed",
          retryable: true,
        };
      }

      return { ok: true };
    },
  };
}

export default function extension(pi: ExtensionAPI): void {
  registerAgentTaskTools(pi, {
    // All tmux panes must run in the same project cwd so they share .pi/tasks.
    store: createFilesystemTaskStore({ tasksDir: ".pi/tasks" }),
    transport: createTmuxTransport(pi),
  });
}

This transport does not open panes. It only pastes assignment text into an already-running pane and presses Enter.

2. open the agents yourself

Start each Pi agent in the same repo/cwd and load the extension. Give each one a stable session name:

# pane 1
PI_TASK_SESSION=parent pi -e ./.pi/extensions/pi-tasks-tmux.ts

# pane 2
PI_TASK_SESSION=worker-a pi -e ./.pi/extensions/pi-tasks-tmux.ts

If the extension is installed with pi install, you can omit -e; the important parts are same cwd, same .pi/tasks, and distinct PI_TASK_SESSION values.

3. find the target pane id

From inside tmux:

tmux list-panes -a -F '#{pane_id} #{pane_current_path} #{pane_title}'

Pane ids look like %12. Use that pane id as agent_task_send.to.

4. send a task

From the parent Pi session, call:

{
  "to": "%12",
  "task": "inspect the auth middleware and report risks",
  "metadata": { "phaseId": "phase-1", "issueId": "auth-review", "role": "reviewer", "verificationTier": "focused" },
  "contextRefs": [{ "path": ".plans/current.md", "required": false, "purpose": "scope" }],
  "timeoutMs": 1800000
}

Expected flow:

parent creates .pi/tasks/<taskId>/...
tmux transport pastes the assignment into pane %12
worker receives pi.task.assignment.v1
worker does the work
worker calls agent_task_done with that taskId
parent reads the result later via agent_task_status or agent_task_inbox; use agent_task_wait only when the current user message explicitly asks to block

tmux troubleshooting

If the target cannot complete the task, confirm it loaded the extension and has agent_task_done available.
If status never changes, confirm parent and worker are in the same cwd and both can see the same .pi/tasks/<taskId> directory.
If dispatch fails, run tmux paste-buffer manually against the same target id; the pane id may be stale.
Do not use terminal output as the source of truth. The task result is the store state, not visible prose in the pane.

using with non-pi workers

A non-Pi worker can participate if it implements the same lifecycle:

receive pi.task.assignment.v1
do the assigned work
write/report a terminal result with the same status and payload semantics
never overwrite an already-terminal task

Example HTTP transport from a Pi extension:

import type { ExtensionAPI } from "@earendil-works/pi-coding-agent";
import { registerAgentTaskTools } from "@sgtbeatdown/pi-tasks/src/extension";
import { createFilesystemTaskStore } from "@sgtbeatdown/pi-tasks/src/stores/filesystem";
import type { TaskTransport } from "@sgtbeatdown/pi-tasks/src/task-communication";

const httpTransport: TaskTransport = {
  name: "http",
  getCurrentSessionName(env) {
    return env.PI_TASK_SESSION ?? "unknown-session";
  },
  async dispatchTask({ target, assignment, task, signal }) {
    const response = await fetch(`https://tasks.example/sessions/${target}/assignments`, {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: JSON.stringify({ taskId: task.id, assignment }),
      signal,
    });

    if (response.ok) return { ok: true };

    return {
      ok: false,
      message: await response.text(),
      retryable: response.status >= 500,
    };
  },
};

export default function extension(pi: ExtensionAPI): void {
  registerAgentTaskTools(pi, {
    store: createFilesystemTaskStore({ tasksDir: ".pi/tasks" }),
    transport: httpTransport,
  });
}

For non-Pi workers, expose a completion endpoint or let workers use your store implementation directly. The worker should report a terminal result equivalent to:

{
  "taskId": "task_...",
  "status": "completed",
  "summary": "finished the assigned work",
  "result": {}
}

file inbox / polling workers

A pushed message is not required. A transport can durably drop assignments into a file inbox or queue and return success once the assignment is written.

One simple design:

store tasks in .pi/tasks
transport writes assignment text to .pi/task-inbox/<target>/<taskId>.txt
worker watches its inbox directory
worker reads the assignment, does the work, then completes through a store API or by invoking a Pi session that has agent_task_done

Completion is separate from dispatch. Dispatch means “assignment delivered,” not “work finished.”

when to implement a custom store

Do not implement a custom store just because you have a new dispatch mechanism. Implement TaskStore only when task state itself must be somewhere else:

central HTTP service
Redis/Postgres/SQLite shared by multiple machines
multi-host queue system
hosted control plane

If only delivery changes, keep the filesystem store.

included wolfpack transport

This package includes a Wolfpack transport for convenience. It is only a delivery adapter:

wolfpack session send <target-session> <assignment>

It does not own storage. The default exported extension uses the generic filesystem store at .pi/tasks/ plus the included Wolfpack transport. Preflight uses structured Wolfpack status only:

wolfpack session status <target-session> --json

Only structured status with terminal.exists === true, terminal.alive === true, and terminal.status === "ready" passes transport_reachable. Structured SESSION_NOT_FOUND, AMBIGUOUS_SELECTOR, and SESSION_DEAD errors fail reachability. BACKEND_UNAVAILABLE, invalid JSON, and command errors are unavailable unless preflight.requireReachable is true. If preflight.requiredProjectDir is supplied, the returned target projectDir or projectPath is compared with resolved paths; missing target project facts fail that required check. Terminal output/prose is never used as the source of truth.

If you are using Wolfpack, install this extension in every participating Pi session and target a session name/id with agent_task_send. Do not use cross-repo agent_task_send with the filesystem store because separate project roots do not share task state. Until a shared/global store exists, direct wolfpack session send is only a fallback instruction channel, not a task completion protocol. Do not use symlinks to imitate shared task storage.

If you are not using Wolfpack, register the tools with your own { store, transport } composition.

tools

tool	purpose
`agent_task_send`	preflight, create, and dispatch a task through the configured transport; optionally set workflow `metadata`, `contextRefs`, `preflight`, and `onCompletePrompt`
`agent_task_status`	read compact structured status for one task, including any sender-defined `onCompletePrompt`
`agent_task_wait`	blocking wait for a terminal result; do not call after dispatch unless the current user message explicitly asks to block
`agent_task_inbox`	list terminal tasks for the current parent session, including any sender-defined `onCompletePrompt`
`agent_task_cancel`	cancel a non-terminal task
`agent_task_done`	assignee-side structured completion; terminates the target response

completion contract

Assigned Pi agents must call agent_task_done as their final action:

{
  "taskId": "task_...",
  "status": "completed",
  "summary": "inspected auth middleware; no bypass found",
  "result": {
    "issueId": "auth-boundary",
    "verdict": "completed",
    "changedFiles": [],
    "verification": [
      { "command": "bun test tests/auth.test.ts", "status": "passed", "exitCode": 0, "summary": "focused auth tests" }
    ],
    "next": "ready for parent review"
  }
}

Keep summary at or below 1200 characters. Put compact machine-readable details under result, using issueId, verdict, changedFiles, verification, blockers, risks, and next when useful. Verification entries should include the exact command, status, exit code, and short summary where applicable.

Non-Pi workers should complete through an equivalent store/server API with the same terminal status and payload semantics.

Valid terminal statuses:

completed
failed
cancelled
timed_out
rejected

Terminal completion is first-writer-wins.

preflight and workflow fields

agent_task_send runs protocol/store preflight before assignment delivery:

empty target strings fail immediately
metadata.issueId blocks duplicate active tasks for the same target + issue
required contextRefs must be project-local and readable
requiredModel and requireIdle currently report unavailable because Pi does not expose target facts yet
transports may implement preflightTarget; otherwise liveness is unavailable, or failed when preflight.requireReachable is true
preflight.requiredProjectDir checks target project facts returned by transport preflight, not the parent cwd

Failed preflight without idempotencyKey returns an ephemeral rejected tool result: no task directory is created and assignment text is not delivered. Failed preflight with idempotencyKey creates or reuses a durable terminal rejected task with error.code = "preflight_failed", saved preflight checks, no assignmentRef, and no delivered assignment text.

contextRefs.selector is stored and sent as opaque metadata in v1. The sender validates only path containment/readability, not line ranges, headings, or JSON pointers.

task protocol

Assignments are delivered as human-readable text plus a structured envelope with:

type: "pi.task.assignment.v1"
taskId
fromSession
instructions
optional workflow metadata
optional bounded contextRefs that the assignee can read explicitly
protocol requirements telling the assignee to use agent_task_done

Task state includes:

pending
dispatched
running
terminal status
assignment/result refs
event history
optional preflight checks/result
optional workflow metadata and context refs
optional sender-defined onCompletePrompt for parent follow-up
optional result payload/artifacts/error

Filesystem layout for dispatched tasks:

<tasksDir>/<taskId>/
├── task.json
├── assignment.json
├── events.jsonl
└── result.json

Durable idempotent preflight-rejected tasks omit assignment.json and have assignmentRef: undefined because no assignment was delivered. Non-idempotent preflight failures are ephemeral and create no task directory.

Generic filesystem storage defaults to .pi/tasks/ when used directly.

postmortem metrics

Analyze an existing filesystem task-store root without reading terminal output:

bun run task-metrics /path/to/project/.pi/tasks

The command prints a deterministic JSON report. The production API is also available directly:

import { analyzeTaskStoreMetrics } from "@sgtbeatdown/pi-tasks/src/metrics";

const report = await analyzeTaskStoreMetrics("/path/to/project/.pi/tasks");

The report includes status counts; durable preflight rejections; failed and unavailable preflight checks by check name; grouping counts for phaseId, issueId, and rootCause; and loopsPerIssueId as the raw task count for each issue. Character totals measure taskText, parsed assignment.json values, object-form instructions, result summaries, and compact JSON.stringify sizes for the value under result.result. A largest-first prompt/result list is capped at 10 entries; each result size is its summary plus serialized structured payload.

Verification counts come only from structured result.result.verification[] entries. The report does not infer verification truth from task/result prose. Missing or malformed artifacts produce structured diagnostics and do not stop analysis of other valid task directories.

task board

Print a compact issue/root-cause board from the same structured artifacts:

bun run task-metrics /path/to/project/.pi/tasks --board

Or call the production API:

import { buildTaskBoard } from "@sgtbeatdown/pi-tasks/src/task-board";

const board = await buildTaskBoard("/path/to/project/.pi/tasks");

The JSON groups complete phaseId/issueId/rootCause tuples, puts groups with active tasks first, then sorts by latest update/completion and identifiers. It includes task/status counts, structured verification/blocker/risk counts, and repeated-issue/shared-root-cause indicators. Output defaults to 10 groups and 20 task ids per group.

Legacy tasks with missing grouping metadata are never hidden or assigned inferred metadata. totalTaskRecords, groupedTaskCount, and ungroupedTaskCount reconcile coverage; ungrouped contains status counts, missing-field reason counts, and up to 20 task ids. Missing/malformed artifacts remain in structured diagnostics.

architecture

pi-tasks separates storage from delivery:

interface TaskStore {
  createOrReuseDispatchedTask(...): Promise<...>;
  createOrReusePreflightRejectedTask(...): Promise<...>;
  readTask(...): Promise<...>;
  readTaskResult(...): Promise<...>;
  expireTaskIfOverdue(...): Promise<...>;
  waitForTask(...): Promise<...>;
  listInbox(...): Promise<...>;
  findActiveTaskByIssueId(...): Promise<...>;
  ackTask(...): Promise<...>;
  cancelTask(...): Promise<...>;
  completeTask(...): Promise<...>;
}

interface TaskTransport {
  getCurrentSessionName(env): string;
  preflightTarget?(input): Promise<TaskPreflightResult>;
  dispatchTask(input): Promise<DispatchTaskResult>;
}

Registration composes both:

registerAgentTaskTools(pi, {
  store: createFilesystemTaskStore({ tasksDir: ".pi/tasks" }),
  transport: myTransport,
});

exported building blocks

src/extension.ts
- registerAgentTaskTools
- createDefaultTaskCommunicationLayer
src/task-communication.ts
- TaskStore
- TaskTransport
- TaskCommunicationLayer
- PreflightTargetInput
src/stores/filesystem.ts
- createFilesystemTaskStore
src/transports/wolfpack.ts
- createWolfpackTaskTransport
src/protocol.ts
- buildAssignment
- compactTaskResult
- validateStructuredTaskResult
src/preflight.ts
- runTaskPreflight
src/metrics.ts
- analyzeTaskStoreMetrics
src/task-board.ts
- buildTaskBoard
src/task-artifacts.ts
- readTaskStoreArtifacts
src/store.ts
- low-level filesystem store functions, if you need finer control than createFilesystemTaskStore

development

bun install
bun test
bun run typecheck

current limitations

no built-in HTTP/Redis transport yet
no runtime transport selector yet; compose a store/transport in an extension
default extension uses the generic filesystem store plus the included Wolfpack transport
Wolfpack preflight depends on the structured wolfpack session status <target> --json contract
the task board is a bounded read-only summary, not a shared registry or project-management system
local development install is pi install /Users/home/Dev/wolfpack-pi-tasks; published releases can use pi install npm:@sgtbeatdown/pi-tasks