diff --git a/packages/pipeline-crew-mcp/src/crew/channel-server.socket.test.ts b/packages/pipeline-crew-mcp/src/crew/channel-server.socket.test.ts index 953fc4df0..43f1bf807 100644 --- a/packages/pipeline-crew-mcp/src/crew/channel-server.socket.test.ts +++ b/packages/pipeline-crew-mcp/src/crew/channel-server.socket.test.ts @@ -78,7 +78,7 @@ describe("crew/channel-server — REAL unix-socket inbox transport", () => { messageId: "m-roundtrip-1", from: "inbox://engineering-manager/1", kind: "IntakePing", - body: {issue: "3489", from: "engineering-manager", at: new Date().toISOString()}, + body: {issue: 3489, from: "engineering-manager", at: new Date().toISOString()}, at: new Date().toISOString(), }; const dialer = yield* Dialer; diff --git a/packages/pipeline-crew-mcp/src/crew/channel-server.test.ts b/packages/pipeline-crew-mcp/src/crew/channel-server.test.ts index 22f990163..4a2f25820 100644 --- a/packages/pipeline-crew-mcp/src/crew/channel-server.test.ts +++ b/packages/pipeline-crew-mcp/src/crew/channel-server.test.ts @@ -125,7 +125,7 @@ describe("crew/channel-server — announce, discover, claim/collision-check (AC Effect.provide(channelLayers(tracker, "inbox://a", Dialer.layerFromConnect(connect))), ); - const ack = yield* a.peer.send(roleB, "IntakePing", {issue: "3059", from: roleA}); + const ack = yield* a.peer.send(roleB, "IntakePing", {issue: 3059, from: roleA}); assert.strictEqual(ack.by, "inbox://b", "acked by B's inbox ⇒ it went straight to B"); const recorded = yield* Ref.get(wakes); assert.lengthOf(recorded, 1); @@ -320,7 +320,7 @@ describe("crew/channel-server — engine pool discovery + per-instance dial (AC messageId: "m-1", from: senderAddr, kind: "IntakePing", - body: {issue: "x"}, + body: {issue: 3059}, at: new Date().toISOString(), }); }).pipe(Effect.provide(channelLayers(tracker, senderAddr, Dialer.layerFromConnect(connect)))); diff --git a/packages/pipeline-crew-mcp/src/crew/contract.test.ts b/packages/pipeline-crew-mcp/src/crew/contract.test.ts new file mode 100644 index 000000000..1b5f9e3cb --- /dev/null +++ b/packages/pipeline-crew-mcp/src/crew/contract.test.ts @@ -0,0 +1,50 @@ +/** + * The crew composition of the discoverable channel contract (#3622): each role's sanctioned seams + * (read off the catalog, so a future non-flat topology stays in sync) joined with the generic + * kind→shape map, and the whole thing resolved as the startup invariant. + */ +import {assert, describe, it} from "@effect/vitest"; +import {Effect} from "effect"; +import {crewMessageKinds} from "../protocol/index.ts"; +import {ALL_SEAMS} from "./catalog.ts"; +import {resolveChannelContract, roleContract, roleContracts} from "./contract.ts"; +import {CREW_ROLES} from "./roles.ts"; + +describe("crew/contract — the discoverable role↔kind surface (#3622)", () => { + it("resolves each role's seams to the wire kinds they ride, off the catalog", () => { + const contract = roleContract("engineering-manager"); + assert.strictEqual(contract.role, "engineering-manager"); + const byName = new Map(contract.seams.map((s) => [s.seam, s.kind])); + // the seam→kind mapping a peer resolves: which wire kind each named seam actually carries. + assert.strictEqual(byName.get("intakePing"), "IntakePing"); + assert.strictEqual(byName.get("claimCollisionCheck"), "Claim"); + assert.strictEqual(byName.get("engineNudge"), "EngineNudge"); + }); + + it("gives every role the full seam set under the flat topology (sourced from the catalog)", () => { + for (const role of CREW_ROLES) { + const seams = roleContract(role).seams.map((s) => s.seam); + assert.deepStrictEqual([...seams], [...ALL_SEAMS]); + } + assert.strictEqual(roleContracts().length, CREW_ROLES.length); + }); + + it.effect("resolveChannelContract yields the full kind set + every role, or fails at boot", () => + Effect.gen(function* () { + const contract = yield* resolveChannelContract(); + assert.deepStrictEqual( + contract.kinds.map((k) => k.kind).sort(), + [...crewMessageKinds].sort(), + ); + assert.deepStrictEqual( + contract.roles.map((r) => r.role), + [...CREW_ROLES], + ); + // the footgun fix is visible in the resolved contract: IntakePing.issue is an integer shape. + const ping = contract.kinds.find((k) => k.kind === "IntakePing"); + const issue = (ping?.payload.schema as {properties?: {issue?: {type?: string}}}).properties + ?.issue; + assert.strictEqual(issue?.type, "integer"); + }), + ); +}); diff --git a/packages/pipeline-crew-mcp/src/crew/contract.ts b/packages/pipeline-crew-mcp/src/crew/contract.ts new file mode 100644 index 000000000..f124c0405 --- /dev/null +++ b/packages/pipeline-crew-mcp/src/crew/contract.ts @@ -0,0 +1,56 @@ +/** + * crew/contract — the crew composition of the discoverable channel contract (#3622): the generic + * kind→shape map (`../protocol/describe`) joined with each role's sanctioned seams (`./catalog`), + * so a peer can resolve BOTH "what does kind K look like" and "which kinds may role R send/receive" + * from one surface, ahead of any send. + * + * The role half is read straight off `seamsFor` / `CrewSeams`, so a future non-flat topology (a role + * mapped to a seam subset instead of `ALL_SEAMS`) flows through automatically — never a hand-kept + * second list. The kind half is `resolveKindContracts`, which fails loud if the shared kind set is + * not fully resolvable (the startup invariant); this module threads that failure up so a session + * refuses to serve an unresolvable contract. + */ +import {Effect} from "effect"; +import { + type ChannelContractError, + type KindContract, + resolveKindContracts, +} from "../protocol/index.ts"; +import {type CrewSeamName, CrewSeams, seamsFor} from "./catalog.ts"; +import {CREW_ROLES, type CrewRole} from "./roles.ts"; + +/** One seam a role is a party to, resolved to the wire kind it rides. */ +export interface SeamContract { + readonly seam: CrewSeamName; + readonly kind: string; +} + +/** A role's sanctioned send/receive seams, sourced from the catalog (so a non-flat topology stays in sync). */ +export interface RoleContract { + readonly role: CrewRole; + readonly seams: ReadonlyArray; +} + +/** The full discoverable channel contract: the shared kind→shape map + each role's sanctioned seams. */ +export interface ChannelContract { + readonly kinds: ReadonlyArray; + readonly roles: ReadonlyArray; +} + +/** The seams `role` is a party to, each resolved to the wire kind it rides (via `CrewSeams`). */ +export const roleContract = (role: CrewRole): RoleContract => ({ + role, + seams: seamsFor(role).map((seam) => ({seam, kind: CrewSeams[seam]._tag})), +}); + +/** Every standing role's seam contract — the role half of the discoverable surface, straight off the catalog. */ +export const roleContracts = (): ReadonlyArray => CREW_ROLES.map(roleContract); + +/** + * Resolve the full discoverable channel contract, or fail loud — the startup invariant (#3622). The + * kind→shape map must resolve for EVERY shared kind before a session serves it; each role's seams are + * read off the catalog. A session wires this on its boot critical path, so an unresolvable kind set + * aborts the build rather than surfacing as a gap at first send. + */ +export const resolveChannelContract = (): Effect.Effect => + resolveKindContracts().pipe(Effect.map((kinds) => ({kinds, roles: roleContracts()}))); diff --git a/packages/pipeline-crew-mcp/src/crew/index.ts b/packages/pipeline-crew-mcp/src/crew/index.ts index 55e889967..d58401c9a 100644 --- a/packages/pipeline-crew-mcp/src/crew/index.ts +++ b/packages/pipeline-crew-mcp/src/crew/index.ts @@ -21,6 +21,14 @@ export { inboxSocketPathFor, makeCrewChannel, } from "./channel-server.ts"; +export { + type ChannelContract, + type RoleContract, + resolveChannelContract, + roleContract, + roleContracts, + type SeamContract, +} from "./contract.ts"; export {RoleUniquenessError} from "./errors.ts"; export { type CardinalityOf, diff --git a/packages/pipeline-crew-mcp/src/crew/session.assembly.test.ts b/packages/pipeline-crew-mcp/src/crew/session.assembly.test.ts index e282ebea0..4761b41f1 100644 --- a/packages/pipeline-crew-mcp/src/crew/session.assembly.test.ts +++ b/packages/pipeline-crew-mcp/src/crew/session.assembly.test.ts @@ -84,7 +84,7 @@ describe("crew/session — forked-stdio live-wake round-trip (session.ts:34 seam messageId: "m-stdio-1", from: "inbox://engineering-manager/1", kind: "IntakePing", - body: {issue: "3489", from: "engineering-manager", at: new Date().toISOString()}, + body: {issue: 3489, from: "engineering-manager", at: new Date().toISOString()}, at: new Date().toISOString(), }) .pipe(Effect.timeout("6 seconds")); diff --git a/packages/pipeline-crew-mcp/src/crew/session.test.ts b/packages/pipeline-crew-mcp/src/crew/session.test.ts index e4ebcaff4..df2dabb9b 100644 --- a/packages/pipeline-crew-mcp/src/crew/session.test.ts +++ b/packages/pipeline-crew-mcp/src/crew/session.test.ts @@ -102,7 +102,7 @@ describe("crew/session — cutover: the ChannelSend-from-peer binding round-trip // The sender's real ChannelSend-from-peer binding (the cutover) — the port channel_send uses. const ack = yield* Effect.gen(function* () { const sender = yield* ChannelSend; - return yield* sender.send(receiverRole, "IntakePing", {issue: "3062", from: senderRole}); + return yield* sender.send(receiverRole, "IntakePing", {issue: 3062, from: senderRole}); }).pipe( Effect.provide( channelSendFromPeer(senderRole, senderAddress).pipe( diff --git a/packages/pipeline-crew-mcp/src/crew/session.ts b/packages/pipeline-crew-mcp/src/crew/session.ts index 6fba7b5ce..4a53b578a 100644 --- a/packages/pipeline-crew-mcp/src/crew/session.ts +++ b/packages/pipeline-crew-mcp/src/crew/session.ts @@ -39,6 +39,7 @@ import {Effect, type FileSystem, Layer} from "effect"; import {type McpSchema, McpServer} from "effect/unstable/ai"; import { ChannelClaim, + ChannelDescribe, ChannelSend, ChannelSink, ChannelToolkit, @@ -46,6 +47,8 @@ import { channelExperimentalCapability, channelToolHandlers, claimToolHandlers, + KindsToolkit, + kindsToolHandlers, } from "../edge/index.ts"; import {type Dialer, Inbox, type Tracker} from "../peer/index.ts"; import {socketPathFor} from "../tracker/index.ts"; @@ -56,6 +59,7 @@ import { inboxSocketPathFor, makeCrewChannel, } from "./channel-server.ts"; +import {resolveChannelContract} from "./contract.ts"; import type {RoleUniquenessError} from "./errors.ts"; import {crewHeartbeatLayer} from "./heartbeat.ts"; import {type CrewRole, kindOf} from "./roles.ts"; @@ -198,35 +202,44 @@ export const assembleCrewSession = ( // Claim FIRST (Race 2): the slow tracker-claim/presence handshake runs in the unwrap's effect, // BEFORE the run-loop-forking transport is built, so ChannelSend below is a zero-async binding. Layer.unwrap( - makeCrewChannel({role: config.role, address}).pipe( - Effect.map((channel) => { - // outbound: the channel_send toolkit, its ChannelSend an INSTANT bind to the already-resolved - // peer — the toolkit registers with no socket await in its build path (Race 2). - const outbound = McpServer.toolkit(ChannelToolkit).pipe( - Layer.provide(channelToolHandlers), - Layer.provide(Layer.succeed(ChannelSend, {send: channel.peer.send})), - ); - // deconfliction: the channel_claim toolkit (#3509), its ChannelClaim an INSTANT bind to the - // already-resolved channel's tracker claim — same zero-async binding as `outbound` (Race 2). - const claim = McpServer.toolkit(ClaimToolkit).pipe( - Layer.provide(claimToolHandlers), - Layer.provide(Layer.succeed(ChannelClaim, {claim: channel.claim})), - ); - // inbound: the peer-inbox socket server, its deliveries waking THIS served server. - const inbound = inboxServerSocketLayer(address).pipe( - Layer.provide(ChannelSink.layerFromMcpServer), - ); - // Provide the transport ONCE to the MERGED registrations — the single-instance memo (Race 1). - return Layer.mergeAll(outbound, claim, inbound).pipe(Layer.provide(transport)); - }), - ), + Effect.gen(function* () { + const channel = yield* makeCrewChannel({role: config.role, address}); + // Startup invariant (#3622): resolve the full discoverable channel contract BEFORE serving. + // A shared kind set that can't be fully resolved to a shape fails the build HERE (on the boot + // critical path, like the claim), so a peer never discovers a gap at first send. + const contract = yield* resolveChannelContract(); + // outbound: the channel_send toolkit, its ChannelSend an INSTANT bind to the already-resolved + // peer — the toolkit registers with no socket await in its build path (Race 2). + const outbound = McpServer.toolkit(ChannelToolkit).pipe( + Layer.provide(channelToolHandlers), + Layer.provide(Layer.succeed(ChannelSend, {send: channel.peer.send})), + ); + // deconfliction: the channel_claim toolkit (#3509), its ChannelClaim an INSTANT bind to the + // already-resolved channel's tracker claim — same zero-async binding as `outbound` (Race 2). + const claim = McpServer.toolkit(ClaimToolkit).pipe( + Layer.provide(claimToolHandlers), + Layer.provide(Layer.succeed(ChannelClaim, {claim: channel.claim})), + ); + // discovery: the channel_kinds toolkit (#3622), its ChannelDescribe an INSTANT bind to the + // contract resolved just above — a static value, so it never re-derives the catalog nor awaits. + const kinds = McpServer.toolkit(KindsToolkit).pipe( + Layer.provide(kindsToolHandlers), + Layer.provide(Layer.succeed(ChannelDescribe, {view: contract})), + ); + // inbound: the peer-inbox socket server, its deliveries waking THIS served server. + const inbound = inboxServerSocketLayer(address).pipe( + Layer.provide(ChannelSink.layerFromMcpServer), + ); + // Provide the transport ONCE to the MERGED registrations — the single-instance memo (Race 1). + return Layer.mergeAll(outbound, claim, kinds, inbound).pipe(Layer.provide(transport)); + }), ).pipe(Layer.provide(substrate)); /** * The full runnable session layer: launch it (`Layer.launch`) to run one live crew session. The one * stdio `McpServer` is provided to the merged registrations exactly once (`assembleCrewSession`), so - * the served server advertises the `channel_send` + `channel_claim` tools + the `claude/channel` - * capability; the heartbeat rides the same `substrate` the outbound binding announces on. + * the served server advertises the `channel_send` + `channel_claim` + `channel_kinds` tools + the + * `claude/channel` capability; the heartbeat rides the same `substrate` the outbound binding announces on. */ export const crewSessionLayer = (config: CrewSessionConfig) => { // One per-session instance id, resolved once here and threaded to every address derivation, so an diff --git a/packages/pipeline-crew-mcp/src/edge/bridge.test.ts b/packages/pipeline-crew-mcp/src/edge/bridge.test.ts index ff00af1d0..76a7ff23b 100644 --- a/packages/pipeline-crew-mcp/src/edge/bridge.test.ts +++ b/packages/pipeline-crew-mcp/src/edge/bridge.test.ts @@ -16,7 +16,7 @@ const envelope = (over?: Partial): InboxEnvelope => ({ messageId: "msg-1", from: "peer-a", kind: "IntakePing", - body: {issue: "3057"}, + body: {issue: 3057}, at: "2026-07-16T10:00:00Z", ...over, }); @@ -47,6 +47,6 @@ describe("edge/bridge — inbound peer-inbox message wakes the session (AC2)", ( it("formatChannelTag renders sender + kind as attributes and the body as content", () => { const tag = formatChannelTag(envelope()); assert.match(tag, /^/); - assert.include(tag, '{"issue":"3057"}'); + assert.include(tag, '{"issue":3057}'); }); }); diff --git a/packages/pipeline-crew-mcp/src/edge/defect-a-double-encode.repro.test.ts b/packages/pipeline-crew-mcp/src/edge/defect-a-double-encode.repro.test.ts index 3b35606e2..8c7c18802 100644 --- a/packages/pipeline-crew-mcp/src/edge/defect-a-double-encode.repro.test.ts +++ b/packages/pipeline-crew-mcp/src/edge/defect-a-double-encode.repro.test.ts @@ -88,9 +88,9 @@ const makeCapturingClient = Effect.gen(function* () { describe("defect(a): channel_send normalizes body to a struct at the boundary (#3491)", () => { it("a STRUCT body renders single-encoded through formatChannelTag (baseline)", () => { - const tag = formatChannelTag(baseEnvelope({issue: "3486", from: "engineering-manager"})); + const tag = formatChannelTag(baseEnvelope({issue: 3486, from: "engineering-manager"})); const inner = tag.slice(tag.indexOf(">") + 1, tag.lastIndexOf("<")); - assert.strictEqual(inner, '{"issue":"3486","from":"engineering-manager"}'); + assert.strictEqual(inner, '{"issue":3486,"from":"engineering-manager"}'); }); // The fix: even when the wire delivers `body` as a JSON STRING, the handler forwards the decoded @@ -105,7 +105,7 @@ describe("defect(a): channel_send normalizes body to a struct at the boundary (# targetRole: "reviewer", kind: "IntakePing", body: JSON.stringify({ - issue: "3486", + issue: 3486, from: "engineering-manager", at: "2026-07-18T00:00:00Z", }), @@ -119,7 +119,7 @@ describe("defect(a): channel_send normalizes body to a struct at the boundary (# "object", "handler must forward the DECODED struct, not the raw JSON string", ); - assert.deepInclude(forwarded as object, {issue: "3486", from: "engineering-manager"}); + assert.deepInclude(forwarded as object, {issue: 3486, from: "engineering-manager"}); const inner = ((tag) => tag.slice(tag.indexOf(">") + 1, tag.lastIndexOf("<")))( formatChannelTag(baseEnvelope(forwarded)), @@ -131,7 +131,7 @@ describe("defect(a): channel_send normalizes body to a struct at the boundary (# ); assert.strictEqual( inner, - '{"issue":"3486","from":"engineering-manager","at":"2026-07-18T00:00:00Z"}', + '{"issue":3486,"from":"engineering-manager","at":"2026-07-18T00:00:00Z"}', ); }), ); diff --git a/packages/pipeline-crew-mcp/src/edge/index.ts b/packages/pipeline-crew-mcp/src/edge/index.ts index d56a5d6c9..f84fcee3d 100644 --- a/packages/pipeline-crew-mcp/src/edge/index.ts +++ b/packages/pipeline-crew-mcp/src/edge/index.ts @@ -19,6 +19,13 @@ export { ClaimToolkit, claimToolHandlers, } from "./claim-tool.ts"; +export { + ChannelContractView, + ChannelDescribe, + DescribeChannelKinds, + KindsToolkit, + kindsToolHandlers, +} from "./kinds-tool.ts"; export { CHANNEL_CAPABILITY, CHANNEL_NOTIFICATION_METHOD, diff --git a/packages/pipeline-crew-mcp/src/edge/kinds-tool.test.ts b/packages/pipeline-crew-mcp/src/edge/kinds-tool.test.ts new file mode 100644 index 000000000..0db1fe4c7 --- /dev/null +++ b/packages/pipeline-crew-mcp/src/edge/kinds-tool.test.ts @@ -0,0 +1,119 @@ +/** + * The discovery tool (#3622): the channel edge server lists a `channel_kinds` tool and a `tools/call` + * returns the resolvable channel contract — every kind's payload shape + each role's sanctioned + * kinds — so a sender reads the shape BEFORE sending instead of triggering a send-time reject. Driven + * against a real in-memory `McpServer.layerHttp` + a session-replaying fetch shim (the same harness + * `send-tool.test.ts` uses). + */ +import {assert, describe, it} from "@effect/vitest"; +import {Effect, Layer, Schema} from "effect"; +import {McpSchema, McpServer} from "effect/unstable/ai"; +import * as FetchHttpClient from "effect/unstable/http/FetchHttpClient"; +import * as HttpRouter from "effect/unstable/http/HttpRouter"; +import {RpcSerialization} from "effect/unstable/rpc"; +import * as RpcClient from "effect/unstable/rpc/RpcClient"; +import { + type ChannelContractView, + ChannelDescribe, + KindsToolkit, + kindsToolHandlers, +} from "./kinds-tool.ts"; +import {channelExperimentalCapability} from "./mcp-channel.ts"; + +class DisposeError extends Schema.TaggedErrorClass()( + "@kampus/pipeline-crew-mcp/DisposeError", + {cause: Schema.Unknown}, +) {} + +// A representative resolved contract the crew composition root would bind — one fire-and-forget kind +// whose `issue` is an integer shape (the footgun fix), and one role's sanctioned seams. +const FIXTURE: ChannelContractView = { + kinds: [ + { + kind: "IntakePing", + awaitsReply: false, + payload: { + schema: {type: "object", properties: {issue: {type: "integer"}}}, + }, + }, + ], + roles: [{role: "engineering-manager", seams: [{seam: "intakePing", kind: "IntakePing"}]}], +}; + +const FakeDescribe = Layer.succeed(ChannelDescribe, {view: FIXTURE}); + +const makeInitializedClient = Effect.gen(function* () { + const serverLayer = McpServer.toolkit(KindsToolkit).pipe( + Layer.provide(kindsToolHandlers), + Layer.provide(FakeDescribe), + Layer.provide( + McpServer.layerHttp({ + name: "ChannelEdgeKindsTestServer", + version: "0.0.0", + path: "/mcp", + experimental: channelExperimentalCapability, + }), + ), + ); + const {dispose, handler} = HttpRouter.toWebHandler(serverLayer, {disableLogger: true}); + yield* Effect.addFinalizer(() => + Effect.tryPromise({try: () => dispose(), catch: (cause) => new DisposeError({cause})}).pipe( + Effect.ignore, + ), + ); + + let sessionId: string | null = null; + const customFetch: typeof fetch = async (input, init) => { + const request = input instanceof Request ? input : new Request(input, init); + if (sessionId) request.headers.set("Mcp-Session-Id", sessionId); + const response = await handler(request); + sessionId = response.headers.get("Mcp-Session-Id"); + return response; + }; + + const clientLayer = RpcClient.layerProtocolHttp({url: "http://localhost/mcp"}).pipe( + Layer.provideMerge([FetchHttpClient.layer, RpcSerialization.layerJsonRpc()]), + Layer.provide(Layer.succeed(FetchHttpClient.Fetch, customFetch)), + ); + const client = yield* RpcClient.make(McpSchema.ClientRpcs).pipe(Effect.provide(clientLayer)); + yield* client.initialize({ + protocolVersion: "9999-01-01", + capabilities: {}, + clientInfo: {name: "TestClient", version: "0.0.0"}, + }); + return {client}; +}); + +describe("edge/kinds-tool — the channel_kinds discovery tool (#3622)", () => { + it.effect("lists the channel_kinds tool", () => + Effect.gen(function* () { + const {client} = yield* makeInitializedClient; + const tools = yield* client["tools/list"]({}); + assert.include( + tools.tools.map((t) => t.name), + "channel_kinds", + ); + }), + ); + + it.effect("channel_kinds returns the resolvable contract (kind shapes + role seams)", () => + Effect.gen(function* () { + const {client} = yield* makeInitializedClient; + const result = yield* client["tools/call"]({name: "channel_kinds", arguments: {}}); + assert.isFalse(result.isError); + const view = result.structuredContent as ChannelContractView; + assert.deepStrictEqual( + view.kinds.map((k) => k.kind), + ["IntakePing"], + ); + // the payload shape a sender resolves ahead of a send — issue as an integer (the footgun fix). + const issue = (view.kinds[0]?.payload as {schema?: {properties?: {issue?: {type?: string}}}}) + .schema?.properties?.issue; + assert.strictEqual(issue?.type, "integer"); + assert.deepStrictEqual( + view.roles.map((r) => r.role), + ["engineering-manager"], + ); + }), + ); +}); diff --git a/packages/pipeline-crew-mcp/src/edge/kinds-tool.ts b/packages/pipeline-crew-mcp/src/edge/kinds-tool.ts new file mode 100644 index 000000000..046da51f9 --- /dev/null +++ b/packages/pipeline-crew-mcp/src/edge/kinds-tool.ts @@ -0,0 +1,70 @@ +/** + * edge/kinds-tool — the DISCOVERY half of the channel edge: an MCP tool a session calls to resolve + * the channel contract BEFORE sending, so a sender reads a kind's payload shape (and which kinds its + * role may send/receive) instead of discovering the shape by triggering a send-time reject (#3622). + * Generic (crew-agnostic); see the boundary note in `../index.ts`. + * + * The contract is INJECTED, exactly like `ChannelSend` / `ChannelClaim`: the edge never builds the + * crew's role topology (the crew composition root does, #3059). The crew resolves the contract on its + * boot critical path — a kind set that can't be fully resolved fails the build (the startup invariant) + * — and binds it here as a static value, so the tool never fails and never re-derives the catalog. + */ +import {Context, Effect, Schema} from "effect"; +import {Tool, Toolkit} from "effect/unstable/ai"; + +/** + * The discoverable channel contract as the tool renders it: every message kind's payload shape (a + * JSON Schema document under `payload`) + each role's sanctioned send/receive seams. Roles and seams + * are opaque strings here — the edge stays crew-agnostic; the crew binds the concrete values. + */ +export const ChannelContractView = Schema.Struct({ + kinds: Schema.Array( + Schema.Struct({ + kind: Schema.String, + awaitsReply: Schema.Boolean, + payload: Schema.Unknown, + }), + ), + roles: Schema.Array( + Schema.Struct({ + role: Schema.String, + seams: Schema.Array(Schema.Struct({seam: Schema.String, kind: Schema.String})), + }), + ), +}); +export type ChannelContractView = typeof ChannelContractView.Type; + +/** + * The describe capability the tool wraps — the contract resolved for THIS session, bound in the crew + * composition root (`../crew/session.ts`). It is a static value (no runtime effect): resolution + * happened at boot, so the tool always succeeds. + */ +export class ChannelDescribe extends Context.Service< + ChannelDescribe, + { + readonly view: ChannelContractView; + } +>()("@kampus/pipeline-crew-mcp/edge/ChannelDescribe") {} + +/** The one describe tool: no arguments in, the full discoverable channel contract out. */ +export const DescribeChannelKinds = Tool.make("channel_kinds", { + description: + "Resolve the crew channel contract BEFORE sending: every message kind's payload shape (JSON " + + "Schema) and each role's sanctioned send/receive kinds. Read this to build a valid `channel_send` " + + "body instead of discovering the shape from a send-time reject.", + parameters: Schema.Struct({}), + success: ChannelContractView, +}); + +/** The describe toolkit — registered on the session's one served `McpServer` alongside `channel_send`. */ +export const KindsToolkit = Toolkit.make(DescribeChannelKinds); + +/** The toolkit handler: return the resolved contract bound via the `ChannelDescribe` port. */ +export const kindsToolHandlers = KindsToolkit.toLayer( + Effect.gen(function* () { + const describer = yield* ChannelDescribe; + return { + channel_kinds: () => Effect.succeed(describer.view), + }; + }), +); diff --git a/packages/pipeline-crew-mcp/src/edge/send-tool.test.ts b/packages/pipeline-crew-mcp/src/edge/send-tool.test.ts index 1ff9f596e..e1952e7c9 100644 --- a/packages/pipeline-crew-mcp/src/edge/send-tool.test.ts +++ b/packages/pipeline-crew-mcp/src/edge/send-tool.test.ts @@ -99,7 +99,7 @@ describe("edge/send-tool — the channel edge server (ACs 1, 3)", () => { arguments: { targetRole: "reviewer", kind: "IntakePing", - body: {issue: "3057", from: "intake", at: "2026-07-16T10:00:00Z"}, + body: {issue: 3057, from: "intake", at: "2026-07-16T10:00:00Z"}, }, }); assert.isFalse(result.isError); @@ -117,7 +117,7 @@ describe("edge/send-tool — the channel edge server (ACs 1, 3)", () => { arguments: { targetRole: "ghost", kind: "IntakePing", - body: {issue: "3057", from: "intake", at: "2026-07-16T10:00:00Z"}, + body: {issue: 3057, from: "intake", at: "2026-07-16T10:00:00Z"}, }, }); assert.isTrue(result.isError); @@ -140,7 +140,7 @@ describe("edge/send-tool — the channel edge server (ACs 1, 3)", () => { targetRole: "ghost", kind: "EngineNudge", body: { - target: {pr: "pr:3649"}, + target: {pr: 3649}, from: "chief-of-staff", at: "2026-07-19T10:00:00Z", }, @@ -163,7 +163,7 @@ describe("edge/send-tool — the channel edge server (ACs 1, 3)", () => { targetRole: "reviewer", kind: "EngineNudge", body: { - target: {issue: "issue:3100"}, + target: {issue: 3100}, from: "chief-of-staff", note: "worth a pass", at: "2026-07-19T10:00:00Z", @@ -185,7 +185,7 @@ describe("edge/send-tool — the channel edge server (ACs 1, 3)", () => { const result = yield* client["tools/call"]({ name: "channel_send", // a typo'd kind (`IntakePng`) that the catalog does not carry - arguments: {targetRole: "reviewer", kind: "IntakePng", body: {issue: "3057"}}, + arguments: {targetRole: "reviewer", kind: "IntakePng", body: {issue: 3057}}, }); assert.isTrue(result.isError); }), @@ -197,7 +197,7 @@ describe("edge/send-tool — the channel edge server (ACs 1, 3)", () => { const result = yield* client["tools/call"]({ name: "channel_send", // a known kind, but the body is missing IntakePing's required `from`/`at` - arguments: {targetRole: "reviewer", kind: "IntakePing", body: {issue: "3057"}}, + arguments: {targetRole: "reviewer", kind: "IntakePing", body: {issue: 3057}}, }); assert.isTrue(result.isError); }), @@ -217,7 +217,7 @@ describe("edge/send-tool — the channel edge server (ACs 1, 3)", () => { targetRole: "reviewer", kind: "IntakePing", // the body as the MCP client sends it for an unconstrained param: a JSON string - body: JSON.stringify({issue: "3057", from: "intake", at: "2026-07-16T10:00:00Z"}), + body: JSON.stringify({issue: 3057, from: "intake", at: "2026-07-16T10:00:00Z"}), }, }); assert.isFalse(result.isError); @@ -236,7 +236,7 @@ describe("edge/send-tool — the channel edge server (ACs 1, 3)", () => { const result = yield* client["tools/call"]({ name: "channel_send", // missing IntakePing's required `from`/`at`, and not a JSON string either - arguments: {targetRole: "reviewer", kind: "IntakePing", body: {issue: "3057"}}, + arguments: {targetRole: "reviewer", kind: "IntakePing", body: {issue: 3057}}, }); assert.isTrue(result.isError); const rendered = (result.content as ReadonlyArray<{text?: string}>) diff --git a/packages/pipeline-crew-mcp/src/peer/inbox.test.ts b/packages/pipeline-crew-mcp/src/peer/inbox.test.ts index bfb819b4c..6a077c8c8 100644 --- a/packages/pipeline-crew-mcp/src/peer/inbox.test.ts +++ b/packages/pipeline-crew-mcp/src/peer/inbox.test.ts @@ -13,7 +13,7 @@ const envelope = (over?: Partial): InboxEnvelope => ({ messageId: "msg-1", from: "peer-a", kind: "IntakePing", - body: {issue: "3056"}, + body: {issue: 3056}, at: "2026-07-16T10:00:00Z", ...over, }); diff --git a/packages/pipeline-crew-mcp/src/peer/peer.test.ts b/packages/pipeline-crew-mcp/src/peer/peer.test.ts index 9092820bf..411022805 100644 --- a/packages/pipeline-crew-mcp/src/peer/peer.test.ts +++ b/packages/pipeline-crew-mcp/src/peer/peer.test.ts @@ -79,7 +79,7 @@ describe("peer/peer — direct peer-to-peer send", () => { const tracker = yield* Tracker; yield* tracker.announce({role: "reviewer", peer: "peer-b", address: "addr-b"}); const a = yield* Peer.make({self: "peer-a", role: "builder", address: "addr-a"}); - const ack = yield* a.send("reviewer", "IntakePing", {issue: "3056"}); + const ack = yield* a.send("reviewer", "IntakePing", {issue: 3056}); // acked BY peer-b's inbox ⇒ it went straight to B, not through the tracker. assert.strictEqual(ack.by, "peer-b"); }).pipe( diff --git a/packages/pipeline-crew-mcp/src/protocol/describe.test.ts b/packages/pipeline-crew-mcp/src/protocol/describe.test.ts new file mode 100644 index 000000000..2ced7b049 --- /dev/null +++ b/packages/pipeline-crew-mcp/src/protocol/describe.test.ts @@ -0,0 +1,76 @@ +/** + * The discoverable kind→shape surface (#3622): a peer resolves each kind's payload shape and the + * whole shared kind set BEFORE sending, instead of discovering a shape by triggering a send-time + * reject. Covers `describeKind` (per-kind shape + reply flag), the full-set resolution, and the + * startup invariant's loud fail on an unresolvable kind. + */ +import {assert, describe, it} from "@effect/vitest"; +import {Effect} from "effect"; +import { + ChannelContractError, + describeKind, + resolveKindContracts, + resolveKindContractsFor, +} from "./describe.ts"; +import {crewMessageKinds} from "./group.ts"; + +describe("protocol/describe — the discoverable kind→shape surface (#3622)", () => { + it.effect("describes every catalog kind, sourced from the same map (no divergent kind set)", () => + Effect.gen(function* () { + for (const kind of crewMessageKinds) { + const contract = yield* describeKind(kind); + assert.isDefined(contract, `every catalog kind is describable: ${kind}`); + assert.strictEqual(contract?.kind, kind); + } + }), + ); + + it.effect( + "surfaces IntakePing.issue as an INTEGER shape ahead of a send (the footgun, now discoverable)", + () => + Effect.gen(function* () { + const contract = yield* describeKind("IntakePing"); + assert.isDefined(contract); + // the payload shape a sender resolves — `issue` is an integer, so `{ "issue": 3621 }` is + // legible as the accepted shape without triggering the send-time reject that hid it (#3622). + const issue = (contract?.payload.schema as {properties?: {issue?: {type?: string}}}) + .properties?.issue; + assert.strictEqual(issue?.type, "integer"); + }), + ); + + it.effect("marks request-response kinds as awaiting a reply, fire-and-forget kinds as not", () => + Effect.gen(function* () { + // Claim + LookupRole carry a typed reply; the rest default to Schema.Void (fire-and-forget). + assert.isTrue((yield* describeKind("Claim"))?.awaitsReply); + assert.isTrue((yield* describeKind("LookupRole"))?.awaitsReply); + assert.isFalse((yield* describeKind("IntakePing"))?.awaitsReply); + assert.isFalse((yield* describeKind("EngineNudge"))?.awaitsReply); + assert.isFalse((yield* describeKind("Heartbeat"))?.awaitsReply); + }), + ); + + it.effect("resolves undefined for a kind outside the catalog", () => + Effect.gen(function* () { + assert.isUndefined(yield* describeKind("NotAKind")); + }), + ); + + it.effect("resolveKindContracts resolves the WHOLE shared kind set (the startup invariant)", () => + Effect.gen(function* () { + const contracts = yield* resolveKindContracts(); + assert.deepStrictEqual(contracts.map((c) => c.kind).sort(), [...crewMessageKinds].sort()); + }), + ); + + it.effect("fails LOUD naming the unresolvable kinds — a gap never waits until first send", () => + Effect.gen(function* () { + const error = yield* resolveKindContractsFor(["IntakePing", "NotAKind", "AlsoMissing"]).pipe( + Effect.flip, + ); + assert.instanceOf(error, ChannelContractError); + assert.deepStrictEqual([...error.unresolved], ["NotAKind", "AlsoMissing"]); + assert.include(error.message, "NotAKind"); + }), + ); +}); diff --git a/packages/pipeline-crew-mcp/src/protocol/describe.ts b/packages/pipeline-crew-mcp/src/protocol/describe.ts new file mode 100644 index 000000000..c919eae65 --- /dev/null +++ b/packages/pipeline-crew-mcp/src/protocol/describe.ts @@ -0,0 +1,111 @@ +/** + * protocol/describe — the DISCOVERABLE half of the typed catalog: render each wire kind to a + * resolvable shape a peer can read BEFORE it sends, instead of discovering the shape only by + * triggering a send-time reject (#3622). The send boundary already enforces the catalog + * (`../edge/send-tool.ts`, #3229); this is the read-ahead surface that enforcement lacked. + * + * Generic (crew-agnostic); see the boundary note in `../index.ts`. Every descriptor is derived + * straight from `CrewProtocol` — the SAME single source `crewMessageKinds` / `payloadSchemaForKind` + * derive from — so the described kind set can never be a second, divergent copy of the catalog. + */ +import {Effect, Schema} from "effect"; +import {CrewProtocol, crewMessageKinds, payloadSchemaForKind} from "./group.ts"; + +/** A kind's resolvable contract: its wire name, whether it awaits a typed reply, and its payload JSON Schema. */ +export interface KindContract { + readonly kind: string; + /** Request-response (awaits a typed reply) vs fire-and-forget (`success` defaults to `Schema.Void`). */ + readonly awaitsReply: boolean; + /** The payload's shape as a JSON Schema document — what a sender resolves to build a valid `body`. */ + readonly payload: ReturnType; +} + +/** + * A boot failure: the shared kind set could not be fully resolved to a describable shape, so a peer + * would discover the gap only at first send. Raised by `resolveKindContracts` and surfaced as the + * startup invariant (#3622) — a crew session refuses to come up rather than serve an unresolvable + * contract. + */ +export class ChannelContractError extends Schema.TaggedErrorClass()( + "@kampus/pipeline-crew-mcp/ChannelContractError", + { + unresolved: Schema.Array(Schema.String), + reason: Schema.String, + }, +) { + override get message(): string { + return this.reason; + } +} + +/** Whether a kind awaits a typed reply — its `success` schema is not the fire-and-forget `Schema.Void` default. */ +const awaitsReplyForKind = (kind: string): boolean => { + const rpc = CrewProtocol.requests.get(kind); + return rpc !== undefined && rpc.successSchema.ast._tag !== "Void"; +}; + +/** + * Describe one kind → its resolvable contract, or `undefined` if the kind is outside the catalog OR + * its payload can't be rendered to a JSON Schema. Effect-typed because the render is the one fallible + * step — `toJsonSchemaDocument` throws on an unrepresentable schema, folded to `undefined` here (never + * a native throw; #2736); the `undefined` is what `resolveKindContractsFor` turns into the loud boot + * failure below. + */ +export const describeKind = (kind: string): Effect.Effect => + Effect.suspend(() => { + const schema = payloadSchemaForKind(kind); + if (schema === undefined) { + return Effect.succeed(undefined); + } + return Effect.try({ + try: () => Schema.toJsonSchemaDocument(schema), + catch: (cause) => String(cause), + }).pipe( + Effect.map( + (payload): KindContract => ({kind, awaitsReply: awaitsReplyForKind(kind), payload}), + ), + Effect.orElseSucceed(() => undefined), + ); + }); + +/** + * Resolve a contract for EVERY kind in `kinds`, or fail loud naming the ones that don't resolve — the + * parametric core of the startup invariant. `resolveKindContracts` binds this to the real catalog; + * exposed so the fail branch (an unresolvable kind ⇒ loud `ChannelContractError`) is directly testable. + */ +export const resolveKindContractsFor = ( + kinds: ReadonlyArray, +): Effect.Effect, ChannelContractError> => + Effect.suspend(() => { + return Effect.gen(function* () { + const contracts: Array = []; + const unresolved: Array = []; + for (const kind of kinds) { + const contract = yield* describeKind(kind); + if (contract === undefined) { + unresolved.push(kind); + } else { + contracts.push(contract); + } + } + if (unresolved.length > 0) { + return yield* Effect.fail( + new ChannelContractError({ + unresolved, + reason: `cannot resolve the shared message-kind contract for: ${unresolved.join(", ")} — refusing to serve a channel whose kind set is not fully discoverable (#3622).`, + }), + ); + } + return contracts; + }); + }); + +/** + * The full kind→shape contract over the shared catalog — or fail loud. This is the startup invariant + * (#3622): a peer that cannot resolve the WHOLE shared kind set fails at boot, so a gap never waits to + * be discovered at first send. Success means every `crewMessageKinds` entry described cleanly. + */ +export const resolveKindContracts = (): Effect.Effect< + ReadonlyArray, + ChannelContractError +> => resolveKindContractsFor(crewMessageKinds); diff --git a/packages/pipeline-crew-mcp/src/protocol/group.test.ts b/packages/pipeline-crew-mcp/src/protocol/group.test.ts index 038f9d75d..d19318265 100644 --- a/packages/pipeline-crew-mcp/src/protocol/group.test.ts +++ b/packages/pipeline-crew-mcp/src/protocol/group.test.ts @@ -63,7 +63,7 @@ describe("protocol/group catalog", () => { it("payloadSchemaForKind resolves a catalog kind to its payload schema (#3229)", () => { // the resolver is derived from the catalog, so every kind resolves and a valid payload decodes assert.strictEqual(payloadSchemaForKind("IntakePing"), Messages.IntakePing); - const ping = {issue: "3057", from: "intake", at: "2026-07-16T10:00:00Z"}; + const ping = {issue: 3057, from: "intake", at: "2026-07-16T10:00:00Z"}; assert.deepStrictEqual( Schema.decodeUnknownSync(payloadSchemaForKind("IntakePing")!)(ping), ping, diff --git a/packages/pipeline-crew-mcp/src/protocol/index.ts b/packages/pipeline-crew-mcp/src/protocol/index.ts index b24d8d929..7f26da65c 100644 --- a/packages/pipeline-crew-mcp/src/protocol/index.ts +++ b/packages/pipeline-crew-mcp/src/protocol/index.ts @@ -4,6 +4,13 @@ * see the boundary note in `../index.ts`. This module defines the wire contract that * tracker, peer, and edge all code against. */ +export { + ChannelContractError, + describeKind, + type KindContract, + resolveKindContracts, + resolveKindContractsFor, +} from "./describe.ts"; export { AnnouncePresence, Claim, diff --git a/packages/pipeline-crew-mcp/src/protocol/schema.test.ts b/packages/pipeline-crew-mcp/src/protocol/schema.test.ts index 5439177e8..7eecbdfef 100644 --- a/packages/pipeline-crew-mcp/src/protocol/schema.test.ts +++ b/packages/pipeline-crew-mcp/src/protocol/schema.test.ts @@ -14,6 +14,11 @@ const roundTrips = >(schema: S, value: S["Type" assert.deepStrictEqual(decoded, value); }; +// Construct branded issue/PR numbers (a bare literal can't satisfy `number & Brand`), the natural +// NUMBER shape the footgun fix mandates (#3622). +const asIssue = Schema.decodeUnknownSync(Messages.IssueNumber); +const asPr = Schema.decodeUnknownSync(Messages.PrNumber); + describe("protocol/schema round-trips", () => { it("kind 1 — claim request + reply", () => { roundTrips(Messages.ClaimRequest, { @@ -52,13 +57,13 @@ describe("protocol/schema round-trips", () => { it("kind 3 — intake ping (with and without the optional note)", () => { roundTrips(Messages.IntakePing, { - issue: "issue:3100", + issue: asIssue(3100), from: "triage", note: "needs a second look", at: "2026-07-16T10:03:00Z", }); roundTrips(Messages.IntakePing, { - issue: "issue:3100", + issue: asIssue(3100), from: "triage", at: "2026-07-16T10:03:00Z", }); @@ -66,13 +71,13 @@ describe("protocol/schema round-trips", () => { it("kind 6 — engine nudge (pr and issue targets, with and without the optional note)", () => { roundTrips(Messages.EngineNudge, { - target: {pr: "pr:3649"}, + target: {pr: asPr(3649)}, from: "chief-of-staff", note: "reviewed + banked, worth a look", at: "2026-07-19T10:03:00Z", }); roundTrips(Messages.EngineNudge, { - target: {issue: "issue:3100"}, + target: {issue: asIssue(3100)}, from: "chief-of-staff", at: "2026-07-19T10:03:00Z", }); diff --git a/packages/pipeline-crew-mcp/src/protocol/schema.ts b/packages/pipeline-crew-mcp/src/protocol/schema.ts index 47a211a51..a301633de 100644 --- a/packages/pipeline-crew-mcp/src/protocol/schema.ts +++ b/packages/pipeline-crew-mcp/src/protocol/schema.ts @@ -21,6 +21,30 @@ export const MessageId = Schema.NonEmptyString; /** An ISO-8601 UTC instant, kept as a string so the wire format stays transport-agnostic + JSON-safe. */ export const Timestamp = Schema.String; +/** + * A GitHub issue number — a positive integer, branded so it can't be confused with a bare count + * or an unrelated string. Modelled as a NUMBER, not a string: a number modelled as a bare string + * is a discoverability footgun — the natural payload `{"issue": 3621}` was rejected `Expected + * string, got 3621` with nothing surfacing the shape ahead of the send (#3622). `PrNumber` is the + * sibling for a pull-request number; both are positive-int-branded so an issue number and a PR + * number are nominally distinct (the union that carries them still keys pr-vs-issue, but the values + * no longer interchange). + */ +export const IssueNumber = Schema.Int.check( + Schema.isGreaterThanOrEqualTo(1, { + title: "IssueNumber", + description: "a GitHub issue number (a positive integer)", + }), +).pipe(Schema.brand("IssueNumber")); + +/** A GitHub pull-request number — a positive integer, branded distinctly from an issue number (see `IssueNumber`, #3622). */ +export const PrNumber = Schema.Int.check( + Schema.isGreaterThanOrEqualTo(1, { + title: "PrNumber", + description: "a GitHub pull-request number (a positive integer)", + }), +).pipe(Schema.brand("PrNumber")); + // Kind 1 — synchronous claim / collision-check (request → typed reply). /** A sender's request to claim a resource; answered by a `ClaimReply` (not fire-and-forget). */ @@ -65,7 +89,7 @@ export const DrainProgressTally = Schema.Struct({ // Kind 3 — intake ping. export const IntakePing = Schema.Struct({ - issue: Schema.NonEmptyString, + issue: IssueNumber, from: RoleId, note: Schema.optionalKey(Schema.String), at: Timestamp, @@ -78,8 +102,8 @@ export const IntakePing = Schema.Struct({ * shape modelled as a union so an ill-formed nudge that names both, or neither, is unrepresentable. */ export const NudgeTarget = Schema.Union([ - Schema.Struct({pr: Schema.NonEmptyString}), - Schema.Struct({issue: Schema.NonEmptyString}), + Schema.Struct({pr: PrNumber}), + Schema.Struct({issue: IssueNumber}), ]); /**