Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
91 changes: 79 additions & 12 deletions apps/server/src/orchestration-v2/ArchitectureArchive.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -19,29 +19,30 @@ import {
nativeWorkerFixture,
} from "../../../../packages/client-runtime/src/testing/nativeWorker.ts";

it.effect(
"reads archived history after closing and reopening the database without a native source",
() =>
it.effect.each([undefined, "table", "consultation"] as const)(
"reads archived history after closing and reopening the database without a native source, mode=%s",
(mode) =>
Effect.gen(function* () {
const fs = yield* FileSystem.FileSystem;
const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-architecture-history-" });
const threadId = ThreadId.make("coordinator");
const lifecycle = {
stage: "waiting",
round: 1,
protocolState: "running",
outcome: "pending",
closure: { state: "pending", members: "pending", director: "pending" },
} as const;
const archive: ArchitectureArchive.ArchitectureArchive = {
summary: {
tableId: "table",
threadId,
providerInstanceId: ProviderInstanceId.make("pi"),
coordinatorSessionId: "native",
roomId: "room",
mode: "table",
...(mode === undefined ? {} : { mode }),
participants: 0,
lifecycle: {
stage: "waiting",
round: 1,
protocolState: "running",
outcome: "pending",
closure: { state: "pending", members: "pending", director: "pending" },
},
lifecycle,
capture: "open",
observation: "stale",
gaps: [],
Expand All @@ -54,13 +55,54 @@ it.effect(
},
providerSessionId: ProviderSessionId.make("session"),
parentProviderThreadId: ProviderThreadId.make("parent"),
records: [],
original: {
id: "original",
text: " # exacto\n🧭 café 中文\n",
source: "original WorkBrief",
complete: true,
},
records: [
{
kind: "room",
id: "room",
...(mode === undefined ? {} : { mode }),
origin: {
coordinatorSessionId: "native",
tool: "architecture_start",
toolCallId: "start",
instanceId: "director",
memberId: "director",
assignmentId: "assignment",
},
goalSource: { kind: "worker-workbrief", instanceId: "director" },
lifecycle,
pendingDoubtIds: [],
},
{
kind: "publication",
id: "composition",
message: {
id: "composition",
ordinal: 1,
roomId: "room",
memberId: "director",
assignmentId: "assignment",
author: "director",
stage: "composition",
snapshotId: "snapshot",
turnId: "turn",
attemptId: "attempt",
native: { sessionId: "session", toolCallId: "submit", persistence: "unconfirmed" },
doubts: [],
body: {
kind: "composition",
snapshotId: "snapshot",
reason: "The director can answer independently",
specialists: [],
},
},
},
],
};
const layer = () => NodeSqliteClient.layer({ filename: `${directory}/history.sqlite` });
yield* Effect.scoped(
Expand All @@ -75,6 +117,31 @@ it.effect(
const sql = yield* SqlClient.SqlClient;
const loaded = yield* ArchitectureArchive.read(sql, threadId);
assert.deepEqual(loaded, [archive]);
const details = yield* ArchitectureQuery.read(
{ threadId, kind: "records", tableId: "table" },
loaded,
[],
);
if (details.kind !== "records") throw Error("Expected records");
if (mode === undefined) {
assert.notProperty(details.summary, "mode");
assert.notProperty(details.items[0], "mode");
} else {
assert.equal(details.summary.mode, mode);
assert.deepInclude(details.items[0], { mode });
}
const composition = yield* ArchitectureQuery.read(
{ threadId, kind: "text", tableId: "table", textId: "publication:composition" },
loaded,
[],
);
if (composition.kind !== "text") throw Error("Expected composition");
assert.deepEqual(JSON.parse(composition.text).body, {
kind: "composition",
snapshotId: "snapshot",
reason: "The director can answer independently",
specialists: [],
});
const result = yield* ArchitectureQuery.read(
{ threadId, kind: "text", tableId: "table", textId: "original" },
loaded,
Expand Down
2 changes: 1 addition & 1 deletion apps/server/src/orchestration-v2/ArchitectureQuery.ts
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ function detail(
return {
kind: record.kind,
id,
mode: record.mode,
...(record.mode === undefined ? {} : { mode: record.mode }),
lifecycle: record.lifecycle,
pendingDoubtIds: record.pendingDoubtIds,
originalTextId: "original",
Expand Down
102 changes: 102 additions & 0 deletions apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -824,6 +824,108 @@ layer("Architecture persisted import", (it) => {
}),
);

it.effect(
"roundtrips mode-free composition and individual mandate without substituting the root brief",
() =>
Effect.gen(function* () {
const f = yield* seedNativeWorker(false);
const a = architecture(f);
const { mode, ...room } = a.room;
assert.equal(mode, "table"); // The other fixtures retain historical modes.
const mandate = "Individual composition mandate 🧭\n".repeat(40);
const specialist: PiArchitectureRecord = {
...a.member,
id: "specialist",
assignmentId: "specialist-assignment",
instanceId: "specialist-instance",
function: "specialist",
mandate: { source: "composition", text: mandate },
compositionMessageId: "composition",
compositionChoiceIndex: 0,
selection: { compositionMessageId: "composition", choiceIndex: 0 },
state: "active",
};
const publication: PiArchitectureRecord = {
kind: "publication",
id: "composition",
message: {
id: "composition",
ordinal: 1,
roomId: room.id,
memberId: a.member.id,
assignmentId: a.member.assignmentId,
author: "director",
stage: "composition",
snapshotId: "snapshot",
turnId: "turn",
attemptId: "attempt",
native: { sessionId: "session", toolCallId: "submit", persistence: "unconfirmed" },
doubts: [],
body: {
kind: "composition",
snapshotId: "snapshot",
reason: "Preserve continuity",
specialists: [{ kind: "retain", memberId: "specialist", mandate, reason: "Needed" }],
},
},
};
const records = [room, a.member, specialist, publication];
yield* f.ingestor.ingestArchitecture({
...a.input,
batch: { ...a.batch, records },
});
const archives = yield* f.sink.architectures(f.parent.threadId);
assert.deepEqual(archives[0]!.records, records);
assert.notProperty(archives[0]!.summary, "mode");
const projections = yield* ProjectionStore.ProjectionStoreV2;
const parent = yield* projections.getThreadProjection(f.parent.threadId);
assert.notProperty(parent.thread.nativeTables![0], "mode");
const child = parent.subagents.find(
(child) => child.nativeWorker?.launch.instanceId === "specialist-instance",
)!;
assert.isAbove(child.prompt.length, 0);
assert.isTrue(mandate.startsWith(child.prompt));
assert.isAtMost(child.prompt.length, 512);
assert.deepEqual(child.nativeWorker?.mandate, {
source: "composition",
text: child.prompt,
complete: false,
});
assert.isUndefined(child.nativeWorker?.brief);
const input = { threadId: f.parent.threadId, tableId: archives[0]!.summary.tableId };
const result = yield* decodeArchitectureReadResult(
yield* queryArchitecture({ ...input, kind: "records" }, archives, parent.subagents),
);
if (result.kind !== "records") throw Error("Expected records");
assert.notProperty(result.summary, "mode");
assert.notProperty(
result.items.find((record) => record.kind === "room")!,
"mode",
);
const detail = result.items.find(
(record) => record.kind === "member" && record.id === "specialist",
);
if (detail?.kind !== "member") throw Error("Expected specialist");
assert.equal(detail.mandateSource, "composition");
for (const [textId, expected] of [
["member:specialist", { mandate: specialist.mandate, bindings: [] }],
[
"publication:composition",
{
body: publication.message.body,
doubts: [],
native: publication.message.native,
},
],
] as const) {
const text = yield* queryArchitecture({ ...input, kind: "text", textId }, archives, []);
if (text.kind !== "text") throw Error("Expected text");
assert.deepEqual(JSON.parse(text.text), expected);
}
assert.deepEqual(archives[0]!.original, a.batch.original);
}),
);

it.effect("queries structured failure attribution and ordered instruction provenance", () =>
Effect.gen(function* () {
const f = yield* seedNativeWorker(false);
Expand Down
4 changes: 2 additions & 2 deletions apps/server/src/orchestration-v2/ProviderEventIngestor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1067,7 +1067,7 @@ export const layer: Layer.Layer<
providerInstanceId: input.providerInstanceId,
coordinatorSessionId,
roomId: input.batch.roomId,
mode: confirmedRoom.mode,
...(confirmedRoom.mode === undefined ? {} : { mode: confirmedRoom.mode }),
participants: members.length,
lifecycle: confirmedRoom.lifecycle,
capture: input.batch.capture,
Expand Down Expand Up @@ -1110,7 +1110,7 @@ export const layer: Layer.Layer<
launch,
roleId: member.roleId,
acceptedAt: member.admittedAt,
...(member.mandate.source === "TeamAssignment.task"
...(member.mandate.source !== "root-workbrief"
? {
mandate: {
source: member.mandate.source,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -948,7 +948,6 @@ it.effect.each(["completed-root", "owned-node", "other-provider-thread", "missin
{
kind: "room",
id: "root-room",
mode: "table",
origin: {
coordinatorSessionId: owner.coordinatorSessionId,
tool: "architecture_start",
Expand Down Expand Up @@ -1061,6 +1060,8 @@ it.effect.each(["completed-root", "owned-node", "other-provider-thread", "missin
assert.equal(archives[0]?.parentProviderThreadId, parent.id);
assert.deepEqual(archives[0]?.records, batch.records);
assert.equal(thread.nativeTables?.[0]?.roomId, batch.roomId);
assert.notProperty(archives[0]!.summary, "mode");
assert.notProperty(thread.nativeTables![0], "mode");
assert.lengthOf(members, 1);
assert.deepEqual(members[0]?.nativeWorker?.launch, {
source: "architecture_roster",
Expand Down
Loading
Loading