/**
 * Synadia Agent Protocol for NATS channel for PI Agent.
 *
 * Implements the clean v0.4 subject scheme (see the sesh-channels SDK
 * `subjects` module). Every PI session becomes a discoverable agent instance
 * (`$SRV.PING/INFO`), addressable on the machine-rooted prompt subject
 * `agents.prompt.<machine>.<project>.<session>`, emitting a periodic
 * heartbeat on `agents.hb.<machine>.<project>.<session>` and answering status
 * requests on `agents.status.<machine>.<project>.<session>`. Agent identity
 * lives in service metadata (the `agent` key), never in a subject.
 *
 * Attachments: inline per spec §5.1/§5.2. Each `{filename, content}` is
 * base64-decoded (strict RFC 4648 §4 — standard alphabet, padded, no
 * whitespace, no URL-safe), the filename is sanitized, bytes are staged on
 * disk at `<STATE_DIR>/attachments/<session>/<uuid>-<filename>`, and their
 * absolute paths are prepended to the prompt text handed to PI. The staging
 * directory is removed on session shutdown. Spec §5.5's artifact endpoint is
 * still the long-term home for large files; this inline path is the
 * small-file story for v0.3.
 */

import { mkdirSync, readFileSync, readdirSync, rmSync, writeFileSync } from "node:fs";
import { homedir } from "node:os";
import { basename, dirname, join } from "node:path";
import { randomUUID } from "node:crypto";
import { readConfig as readRoleClassConfig } from "./config";
import { resolveRawSession } from "./session";
import {
	resolveV04Coord,
	v2EnvelopeText,
	v2Scope,
	V2Stores,
} from "./v04";
import { ulid } from "ulid";

import {
	connect,
	credsAuthenticator,
	headers,
	jwtAuthenticator,
	nkeyAuthenticator,
	tokenAuthenticator,
	usernamePasswordAuthenticator,
	type MsgHdrs,
	type NatsConnection,
} from "@nats-io/transport-node";
import type { NodeConnectionOptions } from "@nats-io/transport-node";
import { Type } from "typebox";
import { Svcm } from "@nats-io/services";
import type { Service, ServiceMsg } from "@nats-io/services";

import {
	SDK_PROTOCOL_VERSION,
	SERVICE_NAME,
	formatHumanBytes,
	parseHumanBytes,
	parseNatsUrl,
} from "@synadia-ai/agents";
import {
	DEFAULT_ATTACHMENTS_OK,
	DEFAULT_MAX_PAYLOAD,
	buildHeartbeatPayload,
	encodeChunk,
	encodeHeartbeatPayload,
	splitResponseText,
} from "@synadia-ai/agent-service";

import { sanitizeSubjectToken, v04 } from "@agent-ops/sesh-channels-sdk";
import type {
	AgentCardPartial,
	Coord,
	Message as V2Message,
	Part as V2Part,
	ScopeRef,
} from "@agent-ops/sesh-channels-sdk";

import type { ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-agent";

// ── Outbound helpers ───────────────────────────────────────────────────────
//
// Mirrors the helper added to claude-nats-channel in PR #6. Converts a plain
// `{string:string}` map into a `MsgHdrs` instance suitable for `nc.publish` /
// `nc.request`. Returns `null` when the input is missing or empty so callers
// can omit the `headers` option entirely.
function buildOutboundHeaders(input: unknown): MsgHdrs | null {
	if (!input || typeof input !== "object") return null;
	const entries = Object.entries(input as Record<string, unknown>);
	if (entries.length === 0) return null;
	const hdrs = headers();
	for (const [k, v] of entries) {
		if (typeof v !== "string") {
			throw new Error(`headers.${k}: value must be a string`);
		}
		hdrs.set(k, v);
	}
	return hdrs;
}

// ─────────────────────────────────────────────────────────────────────────────
// PI-specific protocol constants
// ─────────────────────────────────────────────────────────────────────────────

const SERVICE_VERSION = "0.4.0";

// Heartbeat cadence on `agents.hb.<machine>.<project>.<session>`. Locally
// pinned at 5s so the dashboard's stale-eviction loop (3× intervalS) drops a
// dead `pi` agent in ~15s instead of ~90s. The SDK's
// `DEFAULT_HEARTBEAT_INTERVAL_S` stays at 30s as a sensible third-party
// default — first-party harnesses opt into the snappier cadence.
const HEARTBEAT_INTERVAL_S = 5;

// Spec §2, Appendix C: `pi` is both the canonical agent identifier and its
// conventional subject abbreviation, so `metadata.agent` and the wire
// subject's 3rd token are the same — no `subjectToken` override needed.
const AGENT_ID = "pi";

/** Fallback values used only when `nc.info.max_payload` isn't available.
 *  The live cap comes from the broker after connect — see `maxPayloadBytes`
 *  / `maxPayloadStr` in the extension closure. */
const DEFAULT_MAX_PAYLOAD_BYTES_FALLBACK = parseHumanBytes(DEFAULT_MAX_PAYLOAD);

// Keep-alive `ack` status emitted during long tool runs so the caller's
// inactivity timer (§6.6, typically 60s) doesn't fire before text_delta output.
const ACK_KEEPALIVE_MS = 20_000;

// MOCK_MODE: short-circuit upstream Pi dispatch and answer with a canned
// reply directly from the handler. Used by Slice 7 cross-repo integration
// tests so CI doesn't burn LLM credits on every PR. Production paths
// (MOCK_MODE unset) are entirely unchanged.
const MOCK_MODE = process.env.MOCK_MODE === "1";

// ─────────────────────────────────────────────────────────────────────────────
// Config / paths
// ─────────────────────────────────────────────────────────────────────────────

const STATE_DIR = join(homedir(), ".pi", "agent");
const CONFIG_FILE = join(STATE_DIR, "nats-channel.json");
const ATTACHMENTS_ROOT = join(STATE_DIR, "attachments");
const NATS_CONTEXT_DIR = join(homedir(), ".config", "nats", "context");

// Sesh's live URL files. `~/.sesh/hub.nats.url` is the hub's *client* URL —
// atomically rewritten by `sesh hub serve` (cli/hubinfo.go:writeAtomic) on
// every startup with the freshly-bound ephemeral port, and removed on clean
// shutdown. `<cwd-walk>/.sesh/sessions/<label>.json#nats_url` is the
// per-`sesh up` session's local NATS URL (cli/session.go:Session.Publish,
// removed on Release). Both follow the discovery contract documented in
// sesh's docs/synadia-agents-on-sesh.md and implemented in the ref agent
// at internal/refagent/agent.go:resolveNATSURL — except refagent reads
// `hub.url` (the leaf URL) because it connects as a leaf; we read
// `hub.nats.url` because we connect as a regular NATS client.
const SESH_HUB_NATS_URL_FILE = join(homedir(), ".sesh", "hub.nats.url");
const SESH_SESSIONS_DIRNAME = join(".sesh", "sessions");

const DEFAULT_CONTEXT: NatsContext = {
	url: "demo.nats.io",
	description: "NATS demo server (no auth)",
};

// ─────────────────────────────────────────────────────────────────────────────
// Types
// ─────────────────────────────────────────────────────────────────────────────

type PiNatsConfig = {
	context?: string;
	sessionName?: string;
};

// Matches NATS CLI context files at ~/.config/nats/context/<name>.json.
type NatsContext = {
	description?: string;
	url?: string;
	token?: string;
	user?: string;
	password?: string;
	creds?: string;
	nkey?: string;
	cert?: string;
	key?: string;
	ca?: string;
	tls_first?: boolean;
	inbox_prefix?: string;
	user_jwt?: string;
	user_seed?: string;
	socks_proxy?: string;
};

/** Attachment shape used for staging — distinct from the SDK's
 *  `RequestAttachment` so the staging step deals with already-vetted bytes
 *  under a `bytes` key (legacy pi name). Adapter at the decode boundary. */
type DecodedAttachment = {
	filename: string; // sanitized basename
	bytes: Uint8Array;
};

type PendingRequest = {
	msg: ServiceMsg;
	replySubject: string;
	prompt: string;
	attachments: DecodedAttachment[];
	createdAt: number;
	/** v0.4 routing context. Absent for v1 prompts. */
	v2?: V2RequestContext;
};

/**
 * v0.4 per-request routing context, populated by `handleV2Message`. When
 * present, `message_update` / `agent_end` emit stream events + (best-effort)
 * a final assistant Message on the v0.4 surface instead of the v0.3 reply
 * subject. The v0.3 reply subject is still used for backstop in case the
 * caller is listening there too (rare in practice — v2 callers consume the
 * stream subject).
 */
type V2RequestContext = {
	taskId: string;
	contextId?: string;
	scope: ScopeRef;
	/** Buffered assistant text accumulated across `text_delta` events for the
	 *  final `message` stream event + Message KV append on completion. */
	buffer: string;
	/** Lazy KV/OS opener for this scope. Shared across requests targeting the
	 *  same (scopeKind, scopeId); the parent extension caches one per scope. */
	stores: V2Stores;
	/** Inbound user message recorded at envelope-parse time, persisted to KV
	 *  best-effort on dispatch so observers replaying KV history see the
	 *  trigger. */
	inboundMessage: V2Message;
	/** When the sender set a `reply_to` header (their own prompt subject), the
	 *  final reply is delivered there as a new inbound turn — a valid v2 envelope
	 *  — on `agent_end`, so long-running responses land even after the
	 *  requester's ephemeral inbox timed out (#67). */
	replyToPromptSubject?: string;
};

// ─────────────────────────────────────────────────────────────────────────────
// Pure helpers
// ─────────────────────────────────────────────────────────────────────────────

function loadNatsContext(name: string): NatsContext {
	// Reject names that would escape the context directory. `$NATS_CONTEXT`
	// is set by deployers, not random users, but a clear error beats a
	// stale-file `no 'url' field` message at 3am, and the cost is one
	// guard. Mirrors the validation in
	// `agents/openclaw/src/nats/context-loader.ts`.
	if (
		!name ||
		name.includes("/") ||
		name.includes("\\") ||
		name.includes("\0") ||
		name === ".." ||
		name.startsWith(".")
	) {
		throw new Error(
			`NATS context name ${JSON.stringify(name)} is invalid (must not contain path separators or start with '.')`,
		);
	}
	const contextFile = join(NATS_CONTEXT_DIR, `${name}.json`);
	try {
		return JSON.parse(readFileSync(contextFile, "utf8")) as NatsContext;
	} catch (err) {
		throw new Error(
			`NATS context "${name}" not found at ${contextFile} (${(err as Error).message})`,
		);
	}
}

// ─────────────────────────────────────────────────────────────────────────────
// Sesh live-URL discovery
// ─────────────────────────────────────────────────────────────────────────────
// `sesh hub serve` binds an ephemeral NATS port and writes the URL into
// `~/.sesh/hub.nats.url` atomically. The static NATS CLI context at
// `~/.config/nats/context/<name>.json` does NOT track this — it's a one-shot
// snapshot, so it goes stale on every hub restart. We read the live file at
// startup so the channel follows the current hub without operator
// intervention. If sesh isn't installed, the file is absent and we fall
// through to the static-context path unchanged.

function readTrimmedFile(path: string): string | null {
	try {
		const v = readFileSync(path, "utf8").trim();
		return v || null;
	} catch {
		return null;
	}
}

// Walk from cwd up to root looking for `.sesh/sessions/<label>.json`.
// Mirrors `readSessionNATSURL` in sesh's ref agent at
// `internal/refagent/agent.go:548`.
function readSessionNatsUrl(label: string): string | null {
	let dir = process.cwd();
	while (true) {
		const path = join(dir, SESH_SESSIONS_DIRNAME, `${label}.json`);
		try {
			const data = JSON.parse(readFileSync(path, "utf8")) as { nats_url?: string };
			const url = data.nats_url?.trim();
			return url || null;
		} catch {
			// not present in this dir; walk up
		}
		const parent = dirname(dir);
		if (parent === dir) return null;
		dir = parent;
	}
}

// Find a session label without an explicit `$SESH_SESSION`. Walk cwd → root
// looking for the nearest `.sesh/sessions/` containing exactly one .json
// file (per sesh's lifecycle, a present file = currently-held session;
// release deletes it — see `Session.Release` at cli/session.go:181). If
// the dir has multiple live files, we can't disambiguate and bail.
function discoverSessionLabel(): string | null {
	const explicit = process.env.SESH_SESSION?.trim();
	if (explicit) return explicit;
	let dir = process.cwd();
	while (true) {
		const sessionsDir = join(dir, SESH_SESSIONS_DIRNAME);
		try {
			const files = readdirSync(sessionsDir).filter((f) => f.endsWith(".json"));
			if (files.length === 1) return files[0]!.replace(/\.json$/, "");
			if (files.length > 1) {
				process.stderr.write(
					`pi nats channel: ambiguous sesh sessions in ${sessionsDir}: ${files.join(", ")} — set $SESH_SESSION to pick one; falling through to hub\n`,
				);
				return null;
			}
		} catch {
			// sessions dir doesn't exist here; walk up
		}
		const parent = dirname(dir);
		if (parent === dir) return null;
		dir = parent;
	}
}

function resolveSeshUrl(): { url: string; source: string } | null {
	const label = discoverSessionLabel();
	if (label) {
		const url = readSessionNatsUrl(label);
		if (url) return { url, source: `sesh session "${label}"` };
	}
	const hubUrl = readTrimmedFile(SESH_HUB_NATS_URL_FILE);
	if (hubUrl) return { url: hubUrl, source: "sesh hub (~/.sesh/hub.nats.url)" };
	return null;
}

function contextToConnectOpts(ctx: NatsContext): NodeConnectionOptions {
	const opts: NodeConnectionOptions = { name: "pi-nats-channel" };

	// Parse the URL once; extracted userinfo serves as a fallback only when
	// no explicit context-file auth field is set (precedence below).
	// SDK's `parseNatsUrl` returns `NodeConnectionOptions` with `servers`,
	// plus optional `token`/`user`/`pass` from URL userinfo.
	const urlOpts = ctx.url ? parseNatsUrl(ctx.url) : null;
	if (urlOpts) {
		opts.servers = urlOpts.servers;
	}

	// Auth precedence: explicit context fields > URL userinfo. So a context
	// file with `token: "abc"` wins over `url: "nats://xyz@host:port"`.
	if (ctx.creds) {
		opts.authenticator = credsAuthenticator(readFileSync(ctx.creds));
	} else if (ctx.nkey) {
		opts.authenticator = nkeyAuthenticator(readFileSync(ctx.nkey));
	} else if (ctx.user_jwt && ctx.user_seed) {
		const seed = new TextEncoder().encode(ctx.user_seed);
		opts.authenticator = jwtAuthenticator(ctx.user_jwt, seed);
	} else if (ctx.token) {
		opts.authenticator = tokenAuthenticator(ctx.token);
	} else if (ctx.user) {
		opts.authenticator = usernamePasswordAuthenticator(ctx.user, ctx.password ?? "");
	} else if (urlOpts?.token) {
		opts.authenticator = tokenAuthenticator(urlOpts.token);
	} else if (urlOpts?.user !== undefined) {
		opts.authenticator = usernamePasswordAuthenticator(urlOpts.user, urlOpts.pass ?? "");
	}

	if (ctx.cert || ctx.key || ctx.ca) {
		opts.tls = {
			certFile: ctx.cert || undefined,
			keyFile: ctx.key || undefined,
			caFile: ctx.ca || undefined,
			handshakeFirst: ctx.tls_first || undefined,
		};
	}

	if (ctx.inbox_prefix) opts.inboxPrefix = ctx.inbox_prefix;

	return opts;
}

function loadConfig(): PiNatsConfig {
	try {
		return JSON.parse(readFileSync(CONFIG_FILE, "utf8")) as PiNatsConfig;
	} catch {
		return {};
	}
}

function saveConfig(cfg: PiNatsConfig): void {
	mkdirSync(STATE_DIR, { recursive: true });
	writeFileSync(CONFIG_FILE, `${JSON.stringify(cfg, null, 2)}\n`, "utf8");
}

/**
 * Query existing `agents` service instances and pick the first candidate session
 * name whose `prompt` endpoint subject is free. Auto-suffixes `-2`, `-3`, …
 *
 * Only this owner/agent's subjects can collide with ours (different agent
 * identifiers don't share subjects), so we don't need to filter the discovery
 * response — `taken.has(...)` excludes other namespaces naturally.
 */
async function resolveSessionName(
	nc: NatsConnection,
	base: string,
	owner: string,
): Promise<string> {
	const svcm = new Svcm(nc);
	const client = svcm.client({ strategy: "stall", maxWait: 1000, maxMessages: 50 });

	const taken = new Set<string>();
	try {
		const iter = await client.info(SERVICE_NAME);
		for await (const si of iter) {
			for (const ep of si.endpoints ?? []) {
				taken.add(ep.subject);
			}
		}
	} catch {
		// No existing services or timeout — fine.
	}

	// Probe the clean v0.4 prompt subject for each candidate session name.
	// The prompt subject is session-scoped (5-token), so the dedup probe keys
	// on machine/project/session only — no role token.
	let candidate = base;
	let suffix = 2;
	while (
		taken.has(
			v04.subjects.prompt(
				resolveV04Coord({ owner, sessionName: candidate }),
			),
		)
	) {
		candidate = `${base}-${suffix++}`;
	}
	return candidate;
}

// ─────────────────────────────────────────────────────────────────────────────
// Extension default export
// ─────────────────────────────────────────────────────────────────────────────

export default function (pi: ExtensionAPI) {
	let nc: NatsConnection | undefined;
	let service: Service | undefined;
	let promptSubject: string | undefined;
	let heartbeatSubject: string | undefined;
	let statusSubject: string | undefined;
	let sessionName: string | undefined;
	let owner: string | undefined;
	let instanceId: string | undefined;
	let piCtx: ExtensionContext | undefined;
	let contextLabel: string | undefined;
	let serverUrl: string | undefined;
	// Filled in after connect from `nc.info?.max_payload`; falls back to the
	// SDK's `DEFAULT_MAX_PAYLOAD` (1MB) if the server INFO block is unavailable.
	let maxPayloadBytes = DEFAULT_MAX_PAYLOAD_BYTES_FALLBACK;
	let maxPayloadStr = DEFAULT_MAX_PAYLOAD;

	// v0.4 surfaces — populated alongside the v0.3 ones at service start.
	// `v04Coord` is the resolved {machine, project, session, role[, inst]}
	// used to build the `agents.prompt.*` and `agents.card.*` subjects.
	let v04Coord: Coord | undefined;
	let v04PromptSubject: string | undefined;
	let v04CardSubject: string | undefined;

	let heartbeatTimer: ReturnType<typeof setInterval> | undefined;
	let ackTimer: ReturnType<typeof setInterval> | undefined;

	const pendingRequests = new Map<string, PendingRequest>();
	const requestQueue: string[] = [];
	let activeRequestId: string | null = null;
	let requestCounter = 0;

	// Per-scope `V2Stores` cache. Keyed by `<scopeKind>:<scopeId>` so v2
	// prompts targeting the same scope share a single KV/OS open. First open
	// failure disables the durable path for that scope (the cached entry
	// reports `kv:null,os:null` from `ensure()`); subsequent v2 prompts on
	// other scopes still get a fresh attempt.
	const v2Stores = new Map<string, V2Stores>();
	function getV2Stores(scope: ScopeRef): V2Stores {
		const key = `${scope.scopeKind}:${scope.scopeId}`;
		let s = v2Stores.get(key);
		if (!s) {
			s = new V2Stores(nc!, scope, (msg) => {
				piCtx?.ui.notify(`NATS: ${msg}`, "warning");
			});
			v2Stores.set(key, s);
		}
		return s;
	}

	// Prune pending requests older than 30 min (skip the active one).
	const pruneInterval = setInterval(() => {
		const cutoff = Date.now() - 30 * 60 * 1000;
		for (const [id, req] of pendingRequests) {
			if (id === activeRequestId) continue;
			if (req.createdAt < cutoff) pendingRequests.delete(id);
		}
	}, 60_000);
	pruneInterval.unref();

	// ───────────────────────────────────────────────────────────────────────
	// Chunk publishing (§6)
	// ───────────────────────────────────────────────────────────────────────

	/**
	 * Publish a status chunk on the reply subject (§6.4). Thin wrapper over
	 * the SDK's `encodeChunk` so the call sites stay legible.
	 */
	function publishStatus(replySubject: string, status: string): void {
		if (!nc) return;
		nc.publish(replySubject, encodeChunk({ type: "status", status }));
	}

	/**
	 * Publish response text as one or more `{type:"response",data:<text>}`
	 * chunks (§6.3). Uses the SDK's `splitResponseText` for the UTF-8-safe
	 * split — `reserveBytes: 256` matches the historical pi reserve so chunk
	 * granularity is unchanged from before the SDK migration.
	 */
	function publishResponseText(replySubject: string, text: string): void {
		if (!nc || text.length === 0) return;
		for (const slice of splitResponseText(text, maxPayloadBytes, { reserveBytes: 256 })) {
			nc.publish(replySubject, encodeChunk({ type: "response", text: slice }));
		}
	}

	/**
	 * Publish the end-of-stream terminator: zero-byte body AND no headers
	 * (§6.5). `nc.publish(subject, "")` with no options satisfies both.
	 */
	async function publishTerminator(replySubject: string): Promise<void> {
		if (!nc) return;
		try {
			nc.publish(replySubject, "");
			await nc.flush();
		} catch {
			// best effort — usually disconnecting
		}
	}

	function startAckKeepalive(replySubject: string): void {
		stopAckKeepalive();
		ackTimer = setInterval(() => {
			if (!nc) return;
			try {
				publishStatus(replySubject, "ack");
			} catch {
				// best effort
			}
		}, ACK_KEEPALIVE_MS);
		ackTimer.unref?.();
	}

	function stopAckKeepalive(): void {
		if (ackTimer) {
			clearInterval(ackTimer);
			ackTimer = undefined;
		}
	}

	// ───────────────────────────────────────────────────────────────────────
	// v0.4 stream publishing (per-task `agents.task.stream.*` subject)
	// ───────────────────────────────────────────────────────────────────────

	/**
	 * Emit an incremental `message` stream event carrying a single text-part
	 * Message. Mirrors the v0.3 `publishResponseText` chunker but on the v0.4
	 * stream subject and in the v0.4 wire shape.
	 *
	 * `messageId` is fresh per chunk — v0.4 treats each streamed delta as its
	 * own message rather than overwriting one logical message; the consumer
	 * concatenates parts of all `message` events sharing the same `taskId`
	 * (per A2A semantics).
	 */
	function publishV2TextDelta(v2: V2RequestContext, delta: string): void {
		if (!nc || delta.length === 0) return;
		try {
			v04.stream.publishStreamEvent(
				nc,
				v2.scope,
				v2.taskId,
				v04.stream.streamMessage({
					messageId: `${v2.taskId}.delta.${Date.now()}.${++requestCounter}`,
					role: "agent",
					parts: [{ text: delta }] satisfies V2Part[],
					taskId: v2.taskId,
					contextId: v2.contextId,
				}),
			);
		} catch (e) {
			piCtx?.ui.notify(`NATS v2 publish failed: ${(e as Error).message}`, "warning");
		}
	}

	/**
	 * Emit a `statusUpdate` stream event. `final` MUST be true exactly once
	 * per task (the terminator) — callers in this file flag it on the
	 * COMPLETED/FAILED transition.
	 */
	function publishV2StatusUpdate(
		v2: V2RequestContext,
		state:
			| "TASK_STATE_SUBMITTED"
			| "TASK_STATE_WORKING"
			| "TASK_STATE_COMPLETED"
			| "TASK_STATE_FAILED"
			| "TASK_STATE_CANCELED",
		final: boolean,
		message?: string,
	): void {
		if (!nc) return;
		try {
			v04.stream.publishStreamEvent(
				nc,
				v2.scope,
				v2.taskId,
				v04.stream.streamStatusUpdate({
					taskId: v2.taskId,
					state,
					timestamp: new Date().toISOString(),
					...(message !== undefined ? { message } : {}),
					final,
				}),
			);
		} catch (e) {
			piCtx?.ui.notify(`NATS v2 statusUpdate failed: ${(e as Error).message}`, "warning");
		}
	}

	/**
	 * Emit the terminal assistant `message` stream event carrying the full
	 * concatenated reply text. Distinct from per-delta `publishV2TextDelta`:
	 * consumers that only listen for terminal messages (rather than
	 * subscribing for the whole stream) can pick this one up.
	 *
	 * `messageId` is derived from the task id + a stable `.final` suffix so
	 * the consumer can dedupe.
	 */
	function publishV2FinalMessage(v2: V2RequestContext, text: string): void {
		if (!nc) return;
		try {
			v04.stream.publishStreamEvent(
				nc,
				v2.scope,
				v2.taskId,
				v04.stream.streamMessage({
					messageId: `${v2.taskId}.final`,
					role: "agent",
					parts: [{ text }] satisfies V2Part[],
					taskId: v2.taskId,
					contextId: v2.contextId,
				}),
			);
		} catch (e) {
			piCtx?.ui.notify(`NATS v2 final message failed: ${(e as Error).message}`, "warning");
		}
	}

	// ───────────────────────────────────────────────────────────────────────
	// Queue drain + error reply
	// ───────────────────────────────────────────────────────────────────────

	/**
	 * Respond with a spec §9 error — `respondError` sets the header message,
	 * then we publish the empty terminator per §9.3 / §6.5.
	 */
	function respondWithError(msg: ServiceMsg, code: number, description: string): void {
		try {
			msg.respondError(code, description);
		} catch (e) {
			piCtx?.ui.notify(`NATS: respondError failed: ${(e as Error).message}`, "warning");
		}
		if (msg.reply && nc) {
			try {
				nc.publish(msg.reply, "");
			} catch {
				// best effort
			}
		}
	}

	// ───────────────────────────────────────────────────────────────────────
	// v0.4 prompt handler (additive — runs alongside the v0.3 prompt path)
	// ───────────────────────────────────────────────────────────────────────
	// Pure helpers (`v2EnvelopeText`, `v2Scope`, `resolveV04Coord`) live in
	// `./v04` for unit-test access.

	/**
	 * Inbound handler for the `agents.prompt.*` subject. Parses the v2
	 * envelope, derives task / scope refs, and enqueues into the SAME pi
	 * pipeline used by v0.3 prompts (single in-flight prompt, FIFO queue).
	 *
	 * The v0.3 path is untouched. The only branch that cares about v2 is in
	 * `message_update` / `agent_end`, which checks `pending.v2` and routes
	 * output to `agents.task.stream.*` instead of the v0.3 reply subject.
	 *
	 * Errors during envelope parsing surface as a NATS micro `respondError`
	 * on the request subject — the v0.4 stream subject is never used for
	 * out-of-band error reporting (per the design doc's StreamResponse
	 * union shape, errors are statusUpdate events with state=TASK_STATE_FAILED;
	 * but if we can't parse the envelope we have no taskId to scope that
	 * event to, so the request-reply error is the only honest channel).
	 */
	function handleV2Message(err: Error | null, msg: ServiceMsg): void {
		if (err) {
			piCtx?.ui.notify(`NATS v2 handler error: ${err.message}`, "error");
			return;
		}
		if (msg.data.byteLength > maxPayloadBytes) {
			respondWithError(msg, 400, `payload exceeds max_payload (${maxPayloadStr})`);
			return;
		}

		let envelope: V2Message;
		try {
			envelope = v04.envelope.parseV2Envelope(msg.data);
		} catch (e) {
			respondWithError(msg, 400, `v2 envelope: ${(e as Error).message}`);
			return;
		}

		const text = v2EnvelopeText(envelope);
		if (!text) {
			respondWithError(msg, 400, "v2 envelope: no text parts to dispatch");
			return;
		}

		// Task ID precedence: explicit envelope.taskId > envelope.messageId.
		// Using messageId as a fallback gives the stream subject a well-formed,
		// stable identifier when the caller didn't allocate a task ahead of
		// the prompt.
		const taskId = envelope.taskId ?? envelope.messageId;
		const scope = v2Scope(envelope, taskId);
		const stores = getV2Stores(scope);
		// Capture the requester's durable reply subject (`reply_to` header) so
		// BOTH the MOCK and live paths deliver the final reply there as a new
		// inbound turn (#67).
		const replyToPromptSubject = v04.reply.captureReplyTo(msg.headers);

		// Best-effort durable record of the inbound user prompt — observers
		// replaying KV history see what triggered the task. JetStream-disabled
		// brokers fall through to a warning (logged once per scope) and the
		// stream-event path continues unimpaired.
		void persistInboundMessage(stores, taskId, envelope);

		// Open the durable Task record (create-if-missing bucket + mint SUBMITTED
		// if absent) and transition it to WORKING. Self-sufficient on a shim-less
		// mesh. Fires for both the real and MOCK_MODE paths.
		void markV04TaskWorking(stores, taskId, envelope.contextId);

		// MOCK_MODE short-circuit for v0.4: skip Pi dispatch and emit the
		// protocol-surface artifacts Slice 7's acceptance criteria require —
		// streaming message chunk, durable assistant Message (best-effort), an
		// Artifact, terminal statusUpdate, reply terminator. Bypasses the
		// queue so MOCK_MODE works without an initialised Pi runtime.
		if (MOCK_MODE && nc) {
			void runMockV2Reply(
				msg.reply ?? "",
				taskId,
				scope,
				envelope.contextId,
				stores,
				replyToPromptSubject,
			).catch(
				(e: unknown) => {
					piCtx?.ui.notify(
						`NATS: MOCK_MODE v2 reply failed: ${(e as Error).message}`,
						"warning",
					);
					if (msg.reply && nc) {
						try { nc.publish(msg.reply, ""); } catch { /* gone */ }
					}
				},
			);
			return;
		}

		const requestId = String(++requestCounter);
		pendingRequests.set(requestId, {
			msg,
			// Reply subject is the v2 prompt's reply inbox — used as a backstop
			// keepalive channel (NATS micro acks delivery via reply). Real v2
			// chunks flow on the stream subject; this reply path stays quiet.
			replySubject: msg.reply ?? "",
			prompt: text,
			attachments: [],
			createdAt: Date.now(),
			v2: {
				taskId,
				contextId: envelope.contextId,
				scope,
				buffer: "",
				stores,
				inboundMessage: envelope,
				replyToPromptSubject,
			},
		});
		requestQueue.push(requestId);
		drainQueue();
	}

	/**
	 * Write a v2 user-message envelope to its messages KV bucket.
	 * `idempotentAppend` swallows duplicates with byte-equal payloads, so
	 * shim-side retries don't double-write the same prompt; non-equal
	 * collisions surface as `MessageIDConflict` (logged as a warning so the
	 * operator sees the divergence, not propagated as a request failure —
	 * the prompt itself is already in-flight by the time we write).
	 */
	async function persistInboundMessage(
		stores: V2Stores,
		taskId: string,
		envelope: V2Message,
	): Promise<void> {
		const { kv } = await stores.ensure();
		if (!kv) return;
		try {
			await v04.messages.idempotentAppend(kv, taskId, envelope);
		} catch (e) {
			piCtx?.ui.notify(
				`NATS: v2 inbound message KV append failed: ${(e as Error).message}`,
				"warning",
			);
		}
	}

	/**
	 * Write the assembled assistant Message to the messages KV bucket on
	 * task completion. Best-effort: a KV write failure does NOT block the
	 * terminal `statusUpdate` stream event — observers consuming only the
	 * stream subject still see EOS, and the warning surfaces in the pi UI.
	 *
	 * `messageId` is derived as `<taskId>.final.<ulid>` so retries of the
	 * same completion are byte-equal-idempotent (the same task can't end
	 * twice with different content under our single-active-request queue).
	 */
	async function persistFinalAssistantMessage(
		v2: V2RequestContext,
	): Promise<void> {
		const { kv } = await v2.stores.ensure();
		if (!kv) return;
		const messageId = `${v2.taskId}.final.${ulid()}`;
		const finalMessage: V2Message = {
			messageId,
			role: "agent",
			parts: [{ text: v2.buffer }],
			taskId: v2.taskId,
			...(v2.contextId !== undefined ? { contextId: v2.contextId } : {}),
		};
		try {
			await v04.messages.idempotentAppend(kv, v2.taskId, finalMessage);
		} catch (e) {
			piCtx?.ui.notify(
				`NATS: v2 final message KV append failed: ${(e as Error).message}`,
				"warning",
			);
		}
	}

	/**
	 * Open the durable Task record (create-if-missing bucket + mint SUBMITTED if
	 * absent, via the SDK helper) then transition it to WORKING. The mint makes
	 * the adapter self-sufficient on a shim-less mesh (#66).
	 */
	async function markV04TaskWorking(
		stores: V2Stores,
		taskId: string,
		contextId?: string,
	): Promise<void> {
		const kvTasks = await stores.openTask(
			taskId,
			contextId !== undefined ? { contextId } : {},
		);
		if (!kvTasks) return;
		try {
			await v04.task.updateTaskStatus(kvTasks, taskId, "TASK_STATE_WORKING");
		} catch (e) {
			piCtx?.ui.notify(
				`NATS: v2 task WORKING update failed: ${(e as Error).message}`,
				"warning",
			);
		}
	}

	/**
	 * Complete the durable Task record: set COMPLETED and embed the assistant
	 * result as an artifact in `.artifacts[]` (what GetTask surfaces).
	 * Best-effort.
	 */
	async function completeV04Task(
		stores: V2Stores,
		taskId: string,
		resultText?: string,
	): Promise<void> {
		// Idempotent re-open: the record was minted at task-open; this returns
		// the existing handle without clobbering the record.
		const kvTasks = await stores.openTask(taskId);
		if (!kvTasks) return;
		const text = resultText && resultText.length > 0 ? resultText : "task completed";
		try {
			await v04.task.completeTask(kvTasks, taskId, {
				timestamp: true,
				artifacts: [
					{
						artifactId: ulid(),
						taskId,
						parts: [{ text }],
						name: "result",
						createdAt: new Date().toISOString(),
					},
				],
			});
		} catch (e) {
			piCtx?.ui.notify(
				`NATS: v2 task COMPLETED update failed: ${(e as Error).message}`,
				"warning",
			);
		}
	}

	/**
	 * MOCK_MODE v0.4 mock reply. Produces the same wire artefacts a real
	 * task would (streamed Message event, final Message in KV, an Artifact
	 * in KV+OS, terminal statusUpdate, reply terminator) without paying for
	 * an LLM call. Slice 7 CI relies on this — see plan
	 * `docs/plans/2026-05-25-slice-7-cross-repo-integration.md`.
	 */
	async function runMockV2Reply(
		replySubject: string,
		taskId: string,
		scope: ScopeRef,
		contextId: string | undefined,
		stores: V2Stores,
		replyToPromptSubject?: string,
	): Promise<void> {
		if (!nc) return;
		const mockText = "[MOCK_MODE] streaming chunk 1";
		// Mid-stream message event.
		try {
			v04.stream.publishStreamEvent(
				nc,
				scope,
				taskId,
				v04.stream.streamMessage({
					messageId: `${taskId}.mock.${ulid()}`,
					role: "agent",
					parts: [{ text: mockText }],
					taskId,
					...(contextId !== undefined ? { contextId } : {}),
				}),
			);
		} catch (e) {
			piCtx?.ui.notify(
				`NATS: MOCK_MODE stream message failed: ${(e as Error).message}`,
				"warning",
			);
		}
		// Final assistant Message (best-effort).
		const { kv } = await stores.ensure();
		if (kv) {
			try {
				await v04.messages.idempotentAppend(kv, taskId, {
					messageId: `${taskId}.final.${ulid()}`,
					role: "agent",
					parts: [{ text: mockText }],
					taskId,
					...(contextId !== undefined ? { contextId } : {}),
				});
			} catch (e) {
				piCtx?.ui.notify(
					`NATS: MOCK_MODE final message KV append failed: ${(e as Error).message}`,
					"warning",
				);
			}
		}
		// Artifact (best-effort).
		const artKv = await stores.artifactsKv();
		const { os } = await stores.ensure();
		if (artKv && os) {
			try {
				await v04.artifacts.putArtifact(artKv, os, scope, taskId, {
					parts: [{ text: "[MOCK_MODE] artifact body" }],
					name: "mock-artifact",
					description: "Slice 7 MOCK_MODE canned artifact",
				});
			} catch (e) {
				piCtx?.ui.notify(
					`NATS: MOCK_MODE putArtifact failed: ${(e as Error).message}`,
					"warning",
				);
			}
		}
		// Terminal statusUpdate.
		try {
			v04.stream.publishStreamEvent(
				nc,
				scope,
				taskId,
				v04.stream.streamStatusUpdate({
					taskId,
					state: "TASK_STATE_COMPLETED",
					timestamp: new Date().toISOString(),
					final: true,
				}),
			);
		} catch (e) {
			piCtx?.ui.notify(
				`NATS: MOCK_MODE statusUpdate failed: ${(e as Error).message}`,
				"warning",
			);
		}
		// Durable Task record: COMPLETED + embedded artifact so GetTask returns a
		// terminal record with the artifact in `.artifacts[]`. Idempotent re-open
		// returns the handle for the already-minted record.
		const tasksKv = await stores.openTask(taskId);
		if (tasksKv) {
			try {
				await v04.task.completeTask(tasksKv, taskId, {
					timestamp: true,
					artifacts: [
						{
							artifactId: ulid(),
							taskId,
							parts: [{ text: "[MOCK_MODE] artifact body" }],
							name: "mock-artifact",
							createdAt: new Date().toISOString(),
						},
					],
				});
			} catch (e) {
				piCtx?.ui.notify(
					`NATS: MOCK_MODE task completeTask failed: ${(e as Error).message}`,
					"warning",
				);
			}
		}
		// Durable reply to the requester's `reply_to` subject as a valid v2
		// envelope (#67) — mirrors the live agent_end path so the mock proves the
		// contract. Lands even when the requester's inbox is dead.
		try {
			v04.reply.deliverDurableReply(nc, replyToPromptSubject, mockText, {
				maxPayloadBytes,
				...(contextId !== undefined ? { contextId } : {}),
			});
		} catch (e) {
			piCtx?.ui.notify(
				`NATS: MOCK_MODE durable reply_to delivery failed: ${(e as Error).message}`,
				"warning",
			);
		}
		// Reply-inbox terminator (§6.5).
		if (replySubject) {
			try {
				nc.publish(replySubject, "");
				await nc.flush();
			} catch {
				/* best effort */
			}
		}
	}

	function drainQueue(): void {
		if (activeRequestId) return;
		if (!piCtx || !piCtx.isIdle()) return;

		while (requestQueue.length > 0) {
			const next = requestQueue.shift()!;
			const pending = pendingRequests.get(next);
			if (!pending) continue;

			// Stage attachments to disk and prepend their absolute paths to the
			// prompt. Staging failures (disk full, permission denied) surface to
			// the caller as 500 so they can distinguish them from validation
			// errors — envelope was well-formed, we just couldn't process it.
			let finalPrompt: string;
			try {
				finalPrompt = stageAttachmentsIntoPrompt(pending.prompt, pending.attachments);
			} catch (e) {
				respondWithError(pending.msg, 500, `attachment staging failed: ${(e as Error).message}`);
				pendingRequests.delete(next);
				continue;
			}

			activeRequestId = next;
			try {
				if (pending.v2) {
					// v0.4 acceptance signal — `statusUpdate` event with state
					// WORKING on the per-task stream subject. No reply-subject
					// keepalive (v2 callers consume the stream subject).
					publishV2StatusUpdate(pending.v2, "TASK_STATE_WORKING", false, "ack");
				} else {
					// Tell the caller work has been accepted — resets their inactivity
					// timer before the first text_delta (§6.4).
					publishStatus(pending.replySubject, "ack");
					startAckKeepalive(pending.replySubject);
				}
				// Wrap the prompt in a <channel> tag so pi can recognize NATS-origin
				// messages and apply the prompt-injection guard / terse-reply skill.
				// Mirrors the marker shape claude-code's channel uses, so a single
				// skill can teach both agents the same convention.
				const wrappedPrompt =
					`<channel source="nats" request_id="${next}" session="${sessionName ?? "?"}" ts="${new Date().toISOString()}">\n` +
					finalPrompt +
					`\n</channel>`;
				pi.sendUserMessage(wrappedPrompt);
				return;
			} catch (e) {
				activeRequestId = null;
				stopAckKeepalive();
				requestQueue.unshift(next);
				piCtx.ui.notify(
					`NATS: deferred injection (${(e as Error).message})`,
					"warning",
				);
				return;
			}
		}
	}

	/**
	 * Write each attachment to `<ATTACHMENTS_ROOT>/<session>/<uuid>-<filename>`
	 * and prepend an "[Attachments]" block to the prompt listing their absolute
	 * paths. Returns the augmented prompt.
	 *
	 * Each call uses a fresh UUID subdir so concurrent-looking callers (shouldn't
	 * happen — we serialize — but defense in depth) can't collide. The parent
	 * `<session>` dir is removed on session_shutdown; we don't clean per-request
	 * because follow-up turns in the same session may still reference the paths.
	 */
	function stageAttachmentsIntoPrompt(
		prompt: string,
		attachments: DecodedAttachment[],
	): string {
		if (attachments.length === 0) return prompt;
		if (!sessionName) throw new Error("session not initialized");
		const reqDir = join(ATTACHMENTS_ROOT, sessionName, randomUUID());
		mkdirSync(reqDir, { recursive: true });
		const paths: string[] = [];
		for (const a of attachments) {
			const target = join(reqDir, a.filename);
			writeFileSync(target, a.bytes);
			paths.push(target);
		}
		const list = paths.map((p) => `- ${p}`).join("\n");
		return `[Attachments available at the following absolute paths]\n${list}\n\n${prompt}`;
	}

	// ───────────────────────────────────────────────────────────────────────
	// Heartbeat (§8)
	// ───────────────────────────────────────────────────────────────────────

	function buildHeartbeatBytes(): Uint8Array {
		if (!owner || !instanceId) {
			throw new Error("heartbeat called before service was registered");
		}
		// Identity (agent/owner) is sourced from metadata, not the subject —
		// pass a structural stand-in to the payload builder.
		return encodeHeartbeatPayload(
			buildHeartbeatPayload(
				{ agent: AGENT_ID, owner } as unknown as Parameters<typeof buildHeartbeatPayload>[0],
				HEARTBEAT_INTERVAL_S,
				instanceId,
				{ session: sessionName },
			),
		);
	}

	/**
	 * §8.7 (v0.3) status endpoint handler. Replies with the same JSON payload
	 * shape as a heartbeat (§8.3), freshly built per request — future PRs can
	 * extend the response with richer agent metadata in one place.
	 */
	function handleStatusRequest(err: Error | null, msg: ServiceMsg): void {
		if (err) {
			piCtx?.ui.notify(`NATS status handler error: ${err.message}`, "error");
			return;
		}
		try {
			msg.respond(buildHeartbeatBytes());
		} catch (e) {
			try {
				msg.respondError(500, `status handler error: ${(e as Error).message}`);
			} catch {
				// connection may already be gone
			}
		}
	}

	function startHeartbeat(): void {
		stopHeartbeat();
		if (!nc || !heartbeatSubject) return;
		const publish = (): void => {
			if (!nc || !heartbeatSubject) return;
			try {
				nc.publish(heartbeatSubject, buildHeartbeatBytes());
			} catch {
				// best effort — status loop surfaces real connectivity issues
			}
		};
		// Emit one immediately so callers discovering us don't wait a full cadence.
		publish();
		heartbeatTimer = setInterval(publish, HEARTBEAT_INTERVAL_S * 1000);
		heartbeatTimer.unref?.();
	}

	function stopHeartbeat(): void {
		if (heartbeatTimer) {
			clearInterval(heartbeatTimer);
			heartbeatTimer = undefined;
		}
	}

	// ───────────────────────────────────────────────────────────────────────
	// Status loop + cleanup
	// ───────────────────────────────────────────────────────────────────────

	async function startStatusLoop(conn: NatsConnection, ctx: ExtensionContext): Promise<void> {
		try {
			for await (const s of conn.status()) {
				switch (s.type) {
					case "disconnect":
						ctx.ui.setStatus("nats", "NATS: reconnecting…");
						ctx.ui.notify(`NATS disconnected from ${s.server} — retrying…`, "warning");
						break;
					case "reconnect":
						if (promptSubject) ctx.ui.setStatus("nats", `NATS: ${promptSubject}`);
						ctx.ui.notify(`NATS reconnected to ${s.server}`, "info");
						break;
					case "error":
						ctx.ui.notify(`NATS error: ${s.error.message}`, "error");
						break;
				}
			}
		} catch {
			// Status iterator ended.
		}
	}

	async function cleanup(): Promise<void> {
		stopHeartbeat();
		stopAckKeepalive();
		if (service) {
			try {
				await service.stop();
			} catch {}
			service = undefined;
		}
		if (nc) {
			try {
				await nc.drain();
			} catch {}
			nc = undefined;
		}
		clearInterval(pruneInterval);
		// Remove the session's staged attachments directory.
		if (sessionName) {
			try {
				rmSync(join(ATTACHMENTS_ROOT, sessionName), { recursive: true, force: true });
			} catch {}
		}
		piCtx?.ui.setStatus("nats", undefined);
	}

	// ───────────────────────────────────────────────────────────────────────
	// Event wiring
	// ───────────────────────────────────────────────────────────────────────

	pi.on("session_start", async (_event, ctx) => {
		piCtx = ctx;
		const config = loadConfig();

		// 1. Resolve NATS context. Resolution order:
		//    1. $NATS_CONTEXT env var (explicit deployer override)
		//    2. $NATS_URL env var (raw URL)
		//    3. Sesh live-state discovery — preferred over the static context
		//       when running under sesh, because `sesh hub serve` rotates
		//       ports on every restart and the static context goes stale:
		//        a. <cwd-walk>/.sesh/sessions/<label>.json#nats_url
		//        b. ~/.sesh/hub.nats.url
		//       When sesh state is found AND config.context is set, the
		//       discovered URL overrides the (stale) context URL but the
		//       context's other fields (creds, nkey, tls) still apply.
		//    4. config-file `context` field (static — for non-sesh deployments)
		//    5. built-in default (demo.nats.io, no auth)
		const envCtxName = process.env.NATS_CONTEXT;
		const envUrl = process.env.NATS_URL;
		let natsCtx: NatsContext;
		let resolvedLabel: string;
		try {
			if (envCtxName) {
				natsCtx = loadNatsContext(envCtxName);
				resolvedLabel = `${envCtxName} ($NATS_CONTEXT)`;
			} else if (envUrl) {
				natsCtx = { url: envUrl, description: "from $NATS_URL" };
				resolvedLabel = "$NATS_URL";
			} else {
				const sesh = resolveSeshUrl();
				if (sesh) {
					const base: NatsContext = config.context ? loadNatsContext(config.context) : {};
					natsCtx = { ...base, url: sesh.url, description: sesh.source };
					resolvedLabel = config.context
						? `${sesh.source} (auth from context: ${config.context})`
						: sesh.source;
				} else if (config.context) {
					natsCtx = loadNatsContext(config.context);
					resolvedLabel = config.context;
				} else {
					natsCtx = DEFAULT_CONTEXT;
					resolvedLabel = "default";
				}
			}
		} catch (e) {
			ctx.ui.notify(`NATS: ${(e as Error).message}`, "error");
			ctx.ui.setStatus("nats", "NATS: disconnected");
			return;
		}
		contextLabel = resolvedLabel;
		serverUrl = natsCtx.url ?? "demo.nats.io";

		// 2. Resolve owner + session base name
		owner = sanitizeSubjectToken(process.env.USER ?? "unknown") || "unknown";
		// `resolveRawSession` honours `$SESH_SESSION` (canonical sesh contract,
		// set by `sesh up --exec` / `orch-spawn` SESH_* exports) before falling
		// through to the legacy `$NATS_SESSION_NAME` operator-override and
		// `config.sessionName` / `basename(cwd)` defaults. See
		// `extensions/session.ts` for the ladder spec.
		const rawSession =
			resolveRawSession({
				env: process.env,
				configSessionName: config.sessionName,
				cwd: ctx.cwd,
				sanitize: sanitizeSubjectToken,
			}) || "pi";

		// 3. Connect to NATS
		try {
			const opts = contextToConnectOpts(natsCtx);
			opts.name = `pi-${owner}`;
			nc = await connect(opts);
			if (nc.info?.max_payload) {
				maxPayloadBytes = nc.info.max_payload;
				maxPayloadStr = formatHumanBytes(maxPayloadBytes);
			}
		} catch (e) {
			ctx.ui.notify(
				`NATS connection failed (${serverUrl}): ${(e as Error).message}`,
				"error",
			);
			ctx.ui.setStatus("nats", "NATS: disconnected");
			nc = undefined;
			return;
		}

		// Role/class env reads (Phase 2 of agent-role-class registration).
		// Validated at boot; invalid SESH_ROLE/SESH_CLASS throws ConfigError
		// before we touch the wire. Defaults: role="worker", class="active".
		// Read before session-name resolution so the collision probe can build
		// the role-tiered clean prompt subject.
		const roleClass = readRoleClassConfig();

		// 4. Collision-detect the session name
		try {
			sessionName = await resolveSessionName(nc, rawSession, owner);
		} catch (e) {
			ctx.ui.notify(
				`NATS: session name resolution failed: ${(e as Error).message}`,
				"error",
			);
			await cleanup();
			return;
		}

		// 5. Register the microservice instance (§3)
		// Build the clean v0.4 coord + subjects. Identity (agent/owner/role)
		// lives in service metadata, never in the subject. All verbs — including
		// prompt — are session-scoped 5-token subjects.
		v04Coord = resolveV04Coord({
			owner,
			sessionName,
		});
		try {
			v04PromptSubject = v04.subjects.prompt(v04Coord);
			v04CardSubject = v04.subjects.card(v04Coord);
			promptSubject = v04PromptSubject;
			heartbeatSubject = v04.subjects.heartbeat(v04Coord);
			statusSubject = v04.subjects.status(v04Coord);
		} catch (e) {
			ctx.ui.notify(
				`NATS: v0.4 subject build failed: ${(e as Error).message}`,
				"error",
			);
			await cleanup();
			return;
		}

		// Pre-encode the AgentCard once — same closure-bytes pattern the SDK's
		// `card.registerAgentCard` uses internally. Pi uses raw `Service` from
		// `@nats-io/services` rather than `AgentService`, so we can't call the
		// SDK helper directly (its signature takes `AgentService`); we mirror
		// its serialization (`JSON.stringify(partial)` → UTF-8 bytes) here.
		const cardPartial: AgentCardPartial = {
			description: "PI Agent — coding agent for the PI runtime",
			skills: [
				{
					id: "pi.code",
					name: "Code with PI",
					description:
						"Iterate on a codebase using PI's editor + tool surface — read files, run commands, edit, review.",
					tags: ["coding", "pi", "agent"],
					inputModes: ["text/plain"],
					outputModes: ["text/plain"],
				},
			],
		};
		const cardBytes = new TextEncoder().encode(JSON.stringify(cardPartial));

		try {
			const svcm = new Svcm(nc);
			service = await svcm.add({
				name: SERVICE_NAME,
				version: SERVICE_VERSION,
				description: `PI agent (${sessionName}) in ${ctx.cwd}`,
				metadata: {
					agent: AGENT_ID,
					owner,
					session: sessionName,
					// Machine/project identity lives ONLY in metadata under the
					// clean scheme — the subject carries no agent/class token.
					machine: v04Coord.machine,
					project_id: v04Coord.project,
					protocol_version: `${SDK_PROTOCOL_VERSION.major}.${SDK_PROTOCOL_VERSION.minor}`,
					// v0.4 capability advertisement. The sesh shim reads these to
					// decide prompt routing.
					"sesh.protocol_version": "0.4",
					"sesh.v04_capabilities": "messages,artifacts,cards",
					// Supplementary — preserved but not spec-normative.
					cwd: ctx.cwd,
					role: roleClass.role,
					class: roleClass.class,
				},
				queue: "",
			});
			// Clean v0.4 prompt — `agents.prompt.<machine>.<project>.<session>`.
			// Load-balanced across replicas via PROMPT_QUEUE_GROUP so the
			// shim's `SendMessage` to a role + N replicas spreads across them.
			service.addEndpoint("prompt", {
				subject: v04PromptSubject,
				queue: v04.subjects.PROMPT_QUEUE_GROUP,
				handler: handleV2Message,
				metadata: {
					max_payload: maxPayloadStr,
					attachments_ok: DEFAULT_ATTACHMENTS_OK ? "true" : "false",
					"sesh.protocol_version": "0.4",
					"sesh.envelope": "v2",
				},
			});
			// §8.7: status request/response endpoint. Replies with a freshly-built
			// §8.3 heartbeat payload on every request — same shape as the periodic
			// heartbeat, different transport (request/response instead of pub/sub).
			service.addEndpoint("status", {
				subject: statusSubject,
				queue: "",
				handler: handleStatusRequest,
			});
			// v0.4 AgentCard L3 — public card only. PI doesn't currently expose
			// auth-gated extended skills; skipping `card-extended` makes the
			// shim fall back to `AuthenticatedExtendedCardNotConfiguredError`
			// for clients that ask for it (per design doc).
			service.addEndpoint("card-get", {
				subject: v04CardSubject,
				queue: "",
				handler: (err, msg) => {
					if (err) return;
					try {
						msg.respond(cardBytes);
					} catch {
						// connection may already be gone
					}
				},
				metadata: {
					"sesh.card.layer": "L3",
					"sesh.card.scope": "public",
				},
			});
			instanceId = service.info().id;
		} catch (e) {
			ctx.ui.notify(`NATS: service registration failed: ${(e as Error).message}`, "error");
			await cleanup();
			return;
		}

		// 6. Start heartbeat only AFTER service registration — so anyone who
		//    discovers us via the beacon can resolve metadata via $SRV.INFO (§8.2).
		startHeartbeat();

		// 7. UI feedback
		ctx.ui.setStatus("nats", `NATS: ${promptSubject}`);
		ctx.ui.notify(
			`Connected to NATS (${serverUrl}) as ${promptSubject}`,
			"info",
		);

		// 8. Monitor connection status
		void startStatusLoop(nc, ctx);
	});

	pi.on("message_update", async (event) => {
		if (!nc || !activeRequestId) return;
		const ame = event.assistantMessageEvent;
		if (ame.type !== "text_delta" || !ame.delta) return;
		const pending = pendingRequests.get(activeRequestId);
		if (!pending) return;
		try {
			if (pending.v2) {
				// v0.4: stream the delta as an incremental `message` event and
				// buffer it for the final assistant message on `agent_end`.
				pending.v2.buffer += ame.delta;
				publishV2TextDelta(pending.v2, ame.delta);
			} else {
				publishResponseText(pending.replySubject, ame.delta);
			}
		} catch (e) {
			piCtx?.ui.notify(`NATS: publish failed: ${(e as Error).message}`, "warning");
		}
	});

	pi.on("agent_end", async () => {
		stopAckKeepalive();
		if (nc && activeRequestId) {
			const pending = pendingRequests.get(activeRequestId);
			if (pending) {
				if (pending.v2) {
					// v0.4 terminal sequence:
					//   1. Persist final assistant Message to KV (best-effort).
					//   2. Publish final `message` stream event with full text.
					//   3. Publish `statusUpdate` with state=COMPLETED & final=true.
					// Late observers subscribing only to the stream subject can
					// pick up either signal; keeping both is robust against
					// missed chunks (publishStreamEvent is fire-and-forget).
					// KV write happens first so a stream-event consumer that
					// races to read KV on `final:true` always finds the entry.
					await persistFinalAssistantMessage(pending.v2);
					publishV2FinalMessage(pending.v2, pending.v2.buffer);
					publishV2StatusUpdate(pending.v2, "TASK_STATE_COMPLETED", true);
					// Durable Task record: COMPLETED + embedded artifact so GetTask
					// returns a terminal record (not the shim's SUBMITTED).
					await completeV04Task(
						pending.v2.stores,
						pending.v2.taskId,
						pending.v2.buffer,
					);
					// Durable reply: when the sender set a `reply_to` header,
					// deliver the final reply there as a valid v2 envelope (role
					// "agent") the requester's prompt handler accepts — lands even
					// after the requester's ephemeral inbox timed out (#67).
					try {
						v04.reply.deliverDurableReply(
							nc,
							pending.v2.replyToPromptSubject,
							pending.v2.buffer,
							{
								maxPayloadBytes,
								...(pending.v2.contextId !== undefined
									? { contextId: pending.v2.contextId }
									: {}),
							},
						);
					} catch (e) {
						piCtx?.ui.notify(
							`NATS v2 durable reply_to delivery failed: ${(e as Error).message}`,
							"warning",
						);
					}
					// Empty terminator on the v0.3-style reply (if present) so
					// any tool that's also listening there sees a clean close.
					if (pending.replySubject) {
						await publishTerminator(pending.replySubject);
					} else {
						await nc.flush().catch(() => {});
					}
				} else {
					await publishTerminator(pending.replySubject);
				}
				pendingRequests.delete(activeRequestId);
			}
			activeRequestId = null;
		}
		drainQueue();
	});

	pi.on("session_shutdown", async () => {
		await cleanup();
	});

	// ───────────────────────────────────────────────────────────────────────
	// Commands
	// ───────────────────────────────────────────────────────────────────────

	pi.registerCommand("nats-status", {
		description: "Show NATS channel status",
		handler: async (_args, ctx) => {
			if (!nc || !promptSubject) {
				ctx.ui.notify("NATS: not connected", "warning");
				return;
			}
			const line = [
				`Server: ${serverUrl} (${contextLabel})`,
				`Subject: ${promptSubject}`,
				v04PromptSubject ? `v2: ${v04PromptSubject}` : null,
				v04CardSubject ? `card: ${v04CardSubject}` : null,
				`Service: ${SERVICE_NAME} v${SERVICE_VERSION}`,
				`Protocol: ${SDK_PROTOCOL_VERSION.major}.${SDK_PROTOCOL_VERSION.minor} (+0.4 advertised)`,
				`Instance: ${instanceId ?? "?"}`,
				`Session: ${sessionName}`,
				`Owner: ${owner}`,
				`Pending: ${pendingRequests.size}`,
				`Queued: ${requestQueue.length}`,
				`Active: ${activeRequestId ?? "none"}`,
			]
				.filter(Boolean)
				.join(" • ");
			ctx.ui.notify(line, "info");
		},
	});

	pi.registerCommand("nats-configure", {
		description:
			"Show or update NATS channel configuration (usage: /nats-configure [ <context> | session <name|clear> ])",
		handler: async (args, ctx) => {
			const current = loadConfig();
			const tokens = args.trim().split(/\s+/).filter(Boolean);

			if (tokens.length === 0) {
				const lines = [
					`Context: ${current.context ?? "(default: demo.nats.io)"}`,
					`Session: ${current.sessionName ?? "(auto from cwd)"}`,
				];
				ctx.ui.notify(`NATS config — ${lines.join(" • ")}`, "info");
				return;
			}

			const next: PiNatsConfig = { ...current };
			let changed = false;

			if (tokens[0] === "session") {
				if (tokens[1] === "clear") {
					delete next.sessionName;
					changed = true;
				} else if (tokens[1]) {
					next.sessionName = sanitizeSubjectToken(tokens[1]);
					changed = true;
				} else {
					ctx.ui.notify("Usage: /nats-configure session <name|clear>", "warning");
					return;
				}
			} else {
				// Treat as a context switch — validate by loading first.
				const newContext = tokens[0];
				try {
					loadNatsContext(newContext);
				} catch (e) {
					ctx.ui.notify(`NATS: ${(e as Error).message}`, "error");
					return;
				}
				next.context = newContext;
				changed = true;
			}

			if (!changed) {
				ctx.ui.notify("NATS: no changes", "info");
				return;
			}

			try {
				saveConfig(next);
			} catch (e) {
				ctx.ui.notify(`NATS: failed to save config: ${(e as Error).message}`, "error");
				return;
			}

			ctx.ui.notify(
				"NATS config updated. Restart PI for changes to take effect.",
				"info",
			);
		},
	});

	// ───────────────────────────────────────────────────────────────────────
	// Outbound LLM tools — `publish` / `request`
	//
	// Mirrors the MCP tool surface added to claude-nats-channel +
	// grok-nats-channel in PR #6, but uses pi's first-party
	// `ExtensionAPI.registerTool` (no separate MCP server needed). Tools are
	// registered eagerly at extension-load time; the `nc` guard inside each
	// `execute()` handles the brief window before `session_start` opens the
	// connection.
	// ───────────────────────────────────────────────────────────────────────

	pi.registerTool({
		name: "publish",
		label: "Publish NATS message",
		description:
			"Publish a NATS message (fire-and-forget) on the hub NATS connection used by this session. Use to send to any subject reachable on the broker — including other sesh sessions at agents.prompt.<machine>.<project>.<session> (the clean v0.4 scheme; harness identity lives in service metadata, not the subject). " +
			"Body is sent verbatim as UTF-8 bytes — publish does NOT auto-wrap. Prompt endpoints REQUIRE a valid v2 envelope `{messageId, role, parts}`; a plain-text body is 400-rejected by SAP responders, and because publish is fire-and-forget that rejection is INVISIBLE to you (no reply). Use the `request` tool (which auto-wraps and surfaces errors) when you need a prompt delivered and confirmed.",
		promptSnippet: "publish(subject, body, headers?) — fire-and-forget NATS publish on the session's hub.",
		parameters: Type.Object({
			subject: Type.String({ description: "NATS subject (dot-delimited, no wildcards)." }),
			body: Type.String({ description: "Message body. Sent as UTF-8 bytes verbatim." }),
			headers: Type.Optional(
				Type.Record(Type.String(), Type.String(), {
					description: "Optional NATS headers (string -> string).",
				}),
			),
		}),
		async execute(_toolCallId, params, _signal, _onUpdate, _ctx) {
			if (!nc) throw new Error("nats channel not connected");
			const bytes = new TextEncoder().encode(params.body);
			if (bytes.byteLength > maxPayloadBytes) {
				throw new Error(
					`body exceeds broker max_payload (${maxPayloadBytes} bytes)`,
				);
			}
			const hdrs = buildOutboundHeaders(params.headers);
			nc.publish(
				params.subject,
				bytes,
				hdrs ? { headers: hdrs } : undefined,
			);
			await nc.flush();
			return {
				content: [
					{
						type: "text",
						text: `published to ${params.subject} (${bytes.byteLength} bytes)`,
					},
				],
				details: { subject: params.subject, bytes: bytes.byteLength },
			};
		},
	});

	pi.registerTool({
		name: "request",
		label: "NATS request/reply",
		description:
			"Send a NATS request on the hub NATS connection and wait for the first reply (timeout_ms, default 120000). Returns the reply payload as text. Streaming multi-chunk replies are not collected — only the first reply chunk is returned. " +
			"Body may be plain text — it is auto-wrapped into a valid v2 envelope; pass a JSON envelope to control messageId/role/parts yourself (a valid envelope is sent unchanged). " +
			"If the responder rejects the request (NATS service error), this surfaces as a tool error carrying the code and message — never a silent empty reply. " +
			"For long-running requests where the responder may take longer than timeout_ms to answer, pass reply_to set to your own prompt subject (agents.prompt.<machine>.<project>.<session>) — the responder will deliver its final reply there as a new inbound turn instead of to the ephemeral inbox.",
		promptSnippet:
			"request(subject, body, timeout_ms?, reply_to?, headers?) — NATS request/reply; plain text auto-wraps, responder errors surface; returns the first reply as text. reply_to routes long replies to your prompt subject.",
		parameters: Type.Object({
			subject: Type.String({ description: "NATS subject to send the request to." }),
			body: Type.String({ description: "Request body. Sent as UTF-8 bytes verbatim." }),
			timeout_ms: Type.Optional(
				Type.Integer({
					description: "Reply timeout in milliseconds (default 120000).",
					default: 120000,
					minimum: 1,
				}),
			),
			reply_to: Type.Optional(
				Type.String({
					description:
						"Optional durable reply subject (your own prompt subject, e.g. agents.prompt.<machine>.<project>.<session>). When set, the responder publishes its final reply here as a new inbound turn — bypassing the ephemeral inbox so replies land even after timeout_ms.",
				}),
			),
			headers: Type.Optional(
				Type.Record(Type.String(), Type.String(), {
					description: "Optional NATS headers (string -> string).",
				}),
			),
		}),
		async execute(_toolCallId, params, signal, _onUpdate, _ctx) {
			if (!nc) throw new Error("nats channel not connected");
			// Merge caller-supplied headers with the reply_to header (reply_to
			// wins over any same-key value so the routing contract is honoured).
			let hdrs = buildOutboundHeaders(params.headers);
			if (typeof params.reply_to === "string" && params.reply_to) {
				if (!hdrs) hdrs = headers();
				hdrs.set("reply_to", params.reply_to);
			}
			const timeoutMs = params.timeout_ms ?? 120000;
			if (signal?.aborted) throw new Error("aborted");
			// requestEnveloped coerces the body to a valid v2 envelope (genuine
			// envelope passes through byte-untouched; plain text auto-wraps —
			// #64), enforces the broker max_payload cap, and surfaces a
			// responder service error as a thrown tool error instead of an
			// empty success (#63). pi is an `agent`.
			const { text, msg } = await v04.request.requestEnveloped(
				nc,
				params.subject,
				params.body,
				{
					role: "agent",
					timeout: timeoutMs,
					maxPayloadBytes,
					...(hdrs ? { headers: hdrs } : {}),
				},
			);
			return {
				content: [{ type: "text", text }],
				details: {
					subject: params.subject,
					replyBytes: msg.data.byteLength,
				},
			};
		},
	});
}