feat: agent-colab — live session-to-session collaboration (on by default) - #1196
igor-susic1 wants to merge 2 commits into
Conversation
Each TUI session binds a loopback A2A v1.0 inbox (agent card, message/send, tasks/get/cancel; mandatory bearer token) and registers in a shared peer registry (<agentDir>/peers, pid-liveness pruning). Adds /colab (pick a running session as a worker), /agent-name (persistent naming), and five agent tools: list_peers, link_peer, unlink_peer, ask_peer (blocking delegation with reply capture), message_peer (fire-and-forget, transport-level ack, optional one-shot idle notice). Delivery is context/cache-economical: peer content enters as small labeled plain-text deltas appended at the conversation tail — never transcripts or JSON — so each session stays one unbroken prefix-cache-friendly token stream. Consent controls (AGENT_COLAB_INBOUND=accept|hold|refuse), abuse caps (size, burst, dedupe, in-flight, self-send refusal), TUI-only, kill-switch AGENT_COLAB=off. 37 tests. Co-Authored-By: Kimchi <noreply@kimchi.dev>
|
You can re-trigger by converting this PR to draft, then opening for review again. |
Kimchi Code Review
Summary📊 Review Score: 82/100 (overall code quality — 0 lowest, 100 highest) 🧪 Tests: yes — Strong coverage: 4 test files covering the A2A protocol handler (auth, size caps, burst/dedupe/in-flight limits, task lifecycle, cancel-frees-slot) including a real-socket round-trip; the registry (liveness pruning, malformed records, 0600 permissions, name persistence); the tools end-to-end against a live peer server; and the full extension lifecycle on a mock pi harness (delivery modes, hold/refuse consent, reply capture, one-shot idle notices, peer restart/re-attach, shutdown cleanup, non-TUI skip). Minor hygiene issue: 🔒 Security concerns found: Low-severity, loopback-mitigated concerns: (1) the HTTP listener buffers the entire request body before the bearer-token and size-cap checks run, so a local process can force unbounded memory allocation despite the documented 200 KB cap; (2) the unauthenticated agent card embeds the session's 📝 Found 12 issue(s). See inline comments for details. What to expectKimchi will analyze the changes in this pull request and post:
The review typically completes within a few minutes. This comment will be updated once the review is ready. Interact with Kimchi
ConfigurationReviews are configured by your organization admin. Powered by Kimchi — AI-powered code review by CAST AI |
![kimchi-review[bot]](https://avatars.githubusercontent.com/in/3223788?s=60&v=4){kind=small}
There was a problem hiding this comment.
📊 Review Score: 82/100 (overall code quality — 0 lowest, 100 highest)
⏱️ Estimated effort to review: 4/5 (1 = trivial, 5 = very complex)
🧪 Tests: yes — Strong coverage: 4 test files covering the A2A protocol handler (auth, size caps, burst/dedupe/in-flight limits, task lifecycle, cancel-frees-slot) including a real-socket round-trip; the registry (liveness pruning, malformed records, 0600 permissions, name persistence); the tools end-to-end against a live peer server; and the full extension lifecycle on a mock pi harness (delivery modes, hold/refuse consent, reply capture, one-shot idle notices, peer restart/re-attach, shutdown cleanup, non-TUI skip). Minor hygiene issue: index.test.ts reassigns dir in beforeEach but afterAll only removes the last temp dir.
🔒 Security concerns found: Low-severity, loopback-mitigated concerns: (1) the HTTP listener buffers the entire request body before the bearer-token and size-cap checks run, so a local process can force unbounded memory allocation despite the documented 200 KB cap; (2) the unauthenticated agent card embeds the session's cwd in description, disclosing local filesystem paths to any process that can reach the port; (3) the self-send guard relies solely on sender-supplied fromName metadata, which is forgeable. The bearer token, 0600 registry records, and 127.0.0.1-only binding are otherwise sound.
📝 Found 12 issue(s). See inline comments for details.
| "Self-contained task or question. The peer cannot see your conversation — include file paths and specifics. Ask for a concise answer with pointers, not dumps.", | ||
| minLength: 1, | ||
| }), | ||
| timeoutMs: Type.Optional(Type.Number({ description: "Max seconds to wait for the reply (default 120)." })), |
There was a problem hiding this comment.
Contradictory unit contract on the ask_peer timeout: the parameter is named timeoutMs but the schema description says "Max seconds to wait for the reply (default 120)", and the implementation multiplies by 1000 via (params.timeoutMs ?? 120) * 1000. Since tool descriptions are load-bearing for the model, an agent reading the name will pass milliseconds (e.g. 30000) and block the tool call for ~8 hours instead of 30 seconds; one reading the description passes seconds correctly. There is also no upper bound.
💡 Suggestion: Rename the parameter to timeoutSeconds (or change the description to milliseconds and drop the * 1000), and clamp to a sane range, e.g. Type.Number({ minimum: 5, maximum: 1800 }) with Math.min(...) in the execute path.
| } | ||
| state.tasks.set(task.id, task) | ||
| state.inflightCount += 1 | ||
| // Settle fail-safe: a hung deliver must not leak an in-flight slot forever. |
There was a problem hiding this comment.
makeTask uses setInterval for the settle fail-safe and never clears it. The timer fires every 10 minutes for the lifetime of the process even after the task settles normally (the guard makes re-fires no-ops, and unref prevents it from blocking exit, but every task still leaks a permanently firing interval).
💡 Suggestion: Use setTimeout instead of setInterval (one-shot is the intended semantics), and call clearTimeout(timer) inside settleTask or store the handle on the task so a normally-settled task releases its timer.
| history: [], | ||
| createdAt: Date.now(), | ||
| } | ||
| state.tasks.set(task.id, task) |
There was a problem hiding this comment.
state.tasks grows unboundedly: state.tasks.set(task.id, task) on every message/send, and entries (including up to 200 KB of reply history each) are never removed. A long-lived TUI session with steady peer traffic accumulates task records forever, and tasks/get can resurrect arbitrarily old entries.
💡 Suggestion: Prune settled tasks — e.g. delete entries where settledAt is older than a TTL (5–10 min) inside the existing checkDedupe-style housekeeping or a periodic sweep — and cap the map size.
| port?: number | ||
| }): Promise<{ port: number; stop: () => Promise<void> }> { | ||
| const state = createA2aState(opts) | ||
| const server: Server = createServer((req, res) => { |
There was a problem hiding this comment.
The HTTP listener accumulates the full request body via req.on("data", ...) with no byte limit; both the bearer-token check and the MAX_TEXT_CHARS + 65_536 cap inside handleA2aRequest run only after the entire body is buffered. Any local process can POST an arbitrarily large body to the inbox port and exhaust session memory, which violates the documented 200 KB abuse-resistance cap.
💡 Suggestion: Track the accumulated byte length in the data handler and call req.destroy() (responding with a 413/error) as soon as the total exceeds MAX_TEXT_CHARS + 65_536, before any auth or parsing occurs.
| const started = await startA2aServer({ card, token, deliver }) | ||
| server = started | ||
| card.url = `http://127.0.0.1:${started.port}/` | ||
|
|
There was a problem hiding this comment.
The session_start handler performs unguarded awaits — startA2aServer({ card, token, deliver }) and registerPeer(registryDir, record) (plus writePeerName earlier). Any failure (bind error, unwritable state dir, ENOSPC) rejects the event handler, producing an unhandled rejection and potentially leaving self/activeCard set while the registry record and server are missing — a half-initialized colab state inside an otherwise healthy TUI session.
💡 Suggestion: Wrap the setup block in try/catch: on failure call await teardown(), reset self/activeCard, and ctx.ui.notify(...) that agent-colab is disabled for the session instead of propagating.
| if (text === undefined || text.length === 0) { | ||
| return rpcError(req.id, ERR_INVALID_PARAMS, "message.send requires message.parts with a text part.") | ||
| } | ||
| const from = extractSender(params) |
There was a problem hiding this comment.
ℹ️🔒 Security
The self-send guard compares only from.name === state.card.name — both values are sender-controlled metadata (a peer can omit fromName, and two sessions can legitimately share a name). Self-messaging is therefore easy to trigger, which weakens the documented "agent-to-agent loops die on their own" guarantee since the remaining defenses (burst/dedupe) only rate-limit rather than stop cross-session ping-pong.
💡 Suggestion: Include the receiver's own sessionId in A2aState and reject when from.sessionId matches it; treat same-name matches as ambiguous rather than silently allowing.
| return state.burst.count <= BURST_MAX | ||
| } | ||
|
|
||
| function checkDedupe(state: A2aState, from: PeerSender, text: string): boolean { |
There was a problem hiding this comment.
ℹ️🐛 Bug
checkDedupe keys on from.name + text. Two distinct anonymous peers (no fromName) sending identical text within DEDUPE_WINDOW_MS collide on the same key, so the second peer's legitimate message is rejected with "Duplicate message within the dedupe window."
💡 Suggestion: Incorporate from.sessionId into the dedupe hash input (falling back to name as now), so dedupe is truly per-sender as the header comment states.
| } | ||
|
|
||
| self = { sessionId, name, token } | ||
| const card: AgentCard = { |
There was a problem hiding this comment.
ℹ️🔒 Security
The agent card is served unauthenticated (GET /.well-known/agent-card.json) and its description embeds the session's cwd (kimchi/pi coding session in ${ctx.cwd}). On multi-user machines any local process can scan loopback ports and harvest session names plus project paths (typically including usernames), information the 0600 registry file otherwise protects.
💡 Suggestion: Consider requiring the bearer token for the card as well (peers already read the token from the shared registry before calling), or drop the cwd from the unauthenticated card and expose it only via authenticated RPC.
| } catch { | ||
| return undefined | ||
| } | ||
| } |
There was a problem hiding this comment.
ℹ️🔀 Concurrency
writePeerName does a non-atomic read-modify-write of names.json; two sessions naming themselves concurrently (both start with /agent-name) can lose one entry, and a torn write is silently swallowed on next read (catch resets to {}). Additionally names.json never prunes entries for sessions that no longer exist, so it grows forever.
💡 Suggestion: Write via a temp file + renameSync and re-read/retry on parse failure; opportunistically drop sessionIds not present among live records during listLivePeers or writePeerName.
| agentColab(harness.pi as never, { registryDir: dir }) | ||
| const ctx = { ...baseCtx(), mode: "rpc" } | ||
| await harness.emit("session_start", { reason: "startup" }, ctx) | ||
| expect(listLivePeers(dir)).toHaveLength(0) |
There was a problem hiding this comment.
ℹ️🧪 Testing
beforeEach creates a fresh mkdtempSync dir but afterAll removes only the most recent one (dir is reassigned each test), leaking a temp directory per test into the OS temp dir on every run.
💡 Suggestion: Track dirs in an array (dirs.push(dir) in beforeEach) and remove all of them in afterAll, or clean the current dir in afterEach.
Single source of truth: drop the vendored src/extensions/agent-colab copy; ship the extension as the pinned dependency pi-agent-colab@github:getkimchi/pi-agent-colab#v0.1.0 and mirror its TypeScript files into <agentDir>/extensions/agent-colab at startup (before extension discovery) via src/integrations/agent-colab.ts — version-stamp gated, best-effort, same write-into-extensions-dir pattern as the herdr bridge. The pi extension loader aliases bare imports (typebox, pi-tui, pi-coding-agent) to bundled copies, so the mirrored files need no node_modules of their own. Co-Authored-By: Kimchi <noreply@kimchi.dev>
2 participants
What
Integrates agent-colab as a built-in, default-on extension: running kimchi TUI sessions discover each other and can delegate bounded work to one another — subagent-style delegation, but the worker is a full interactive session with its own user.
/colab(pick a live session as a worker),/agent-name(persistent naming)list_peers,link_peer/unlink_peer,ask_peer(blocking, reply captured),message_peer(fire-and-forget, optional one-shot idle notice)message/send,tasks/get/cancel, mandatory bearer token) + shared peer registry at<agentDir>/peerswith pid-liveness pruningWhy this design
nextTurnwhen idle (no forced turn); onlyask_peer(sender blocked on the answer) wakes an idle peer.AGENT_COLAB_INBOUND=accept|hold|refuse; peer messages cannot approve permissions, change config, or run commands; receiver-side permission gates still apply; sender labeled in the transcript.Integration notes
pi-agent-colabis a pinneddependency (
github:getkimchi/pi-agent-colab#v0.1.0). At startup — before extensiondiscovery —
src/integrations/agent-colab.tsmirrors its TypeScript files into<agentDir>/extensions/agent-colaband stamps the installed version (samewrite-into-extensions-dir pattern as the herdr bridge; version stamp → no-op on later
starts). pi's extension loader aliases bare imports (
typebox,pi-tui,pi-coding-agent) to its bundled copies, so the mirrored files need no node_modules.Upstream repo owns the code and its 37 tests; kimchi adds an installer test (5 tests).
docs/agent-colab.md.list_peers/colab, new/late/reloaded peers appear on their own, and linked workers re-attach by persisted name after peer restarts.ExtensionAPI,getAgentDir),typebox, andpi-tui; the agent dir for the registry is inferred from the live session-file path, so pi and kimchi installs converge on their own redirected agent dirs.Test plan
pnpm vitest run src/integrations— installer tests passing (extension's own 37 tests run upstream in getkimchi/pi-agent-colab)pnpm exec tsc --noEmitclean;biome checkclean on touched paths/colabpick →ask_peerround-trip → fire-and-forget + idle notice →/reloadattach on a running session