fix: 校正实时探针可选工具事件合同
This commit is contained in:
Codex
@@ -206,6 +206,10 @@ export interface HwlabRuntimeWebProbeRealtimeFanoutProfileSpec {
|
||||
agentrun: readonly string[];
|
||||
hwlab: readonly string[];
|
||||
}>;
|
||||
readonly conditionalEventTypePairs: Readonly<Record<string, Readonly<{
|
||||
agentrun: string;
|
||||
hwlab: string;
|
||||
}>>>;
|
||||
}
|
||||
|
||||
export interface HwlabRuntimeWebProbeAuthLoginSpec {
|
||||
@@ -1514,6 +1518,24 @@ function webProbeRealtimeFanoutProfileConfig(value: unknown, path: string): Hwla
|
||||
const hwlabEventTypes = nonEmptyStringArrayField(requiredEventTypes, "hwlab", `${path}.requiredEventTypes`);
|
||||
if (new Set(agentrunEventTypes).size !== agentrunEventTypes.length) throw new Error(`${path}.requiredEventTypes.agentrun must not contain duplicates`);
|
||||
if (new Set(hwlabEventTypes).size !== hwlabEventTypes.length) throw new Error(`${path}.requiredEventTypes.hwlab must not contain duplicates`);
|
||||
const conditionalEventTypePairsRaw = asRecord(raw.conditionalEventTypePairs, `${path}.conditionalEventTypePairs`);
|
||||
if (Object.keys(conditionalEventTypePairsRaw).length === 0) throw new Error(`${path}.conditionalEventTypePairs must contain at least one pair`);
|
||||
const conditionalEventTypePairs = Object.fromEntries(Object.entries(conditionalEventTypePairsRaw).map(([id, value]) => {
|
||||
if (!/^[a-z][a-z0-9-]*$/u.test(id)) throw new Error(`${path}.conditionalEventTypePairs.${id} must use a stable lowercase id`);
|
||||
const pairPath = `${path}.conditionalEventTypePairs.${id}`;
|
||||
const pair = asRecord(value, pairPath);
|
||||
const unknownFields = Object.keys(pair).filter((field) => field !== "agentrun" && field !== "hwlab");
|
||||
if (unknownFields.length > 0) throw new Error(`${pairPath}.${unknownFields[0]} is not allowed`);
|
||||
const agentrun = stringField(pair, "agentrun", pairPath);
|
||||
const hwlab = stringField(pair, "hwlab", pairPath);
|
||||
if (agentrunEventTypes.includes(agentrun)) throw new Error(`${pairPath}.agentrun duplicates a required event type`);
|
||||
if (hwlabEventTypes.includes(hwlab)) throw new Error(`${pairPath}.hwlab duplicates a required event type`);
|
||||
return [id, { agentrun, hwlab }];
|
||||
}));
|
||||
const conditionalAgentrunTypes = Object.values(conditionalEventTypePairs).map((pair) => pair.agentrun);
|
||||
const conditionalHwlabTypes = Object.values(conditionalEventTypePairs).map((pair) => pair.hwlab);
|
||||
if (new Set(conditionalAgentrunTypes).size !== conditionalAgentrunTypes.length) throw new Error(`${path}.conditionalEventTypePairs agentrun types must not contain duplicates`);
|
||||
if (new Set(conditionalHwlabTypes).size !== conditionalHwlabTypes.length) throw new Error(`${path}.conditionalEventTypePairs hwlab types must not contain duplicates`);
|
||||
return {
|
||||
subscriberCount: 2,
|
||||
debugStreams: debugStreams as ("stdio" | "agentrun" | "hwlab")[],
|
||||
@@ -1535,6 +1557,7 @@ function webProbeRealtimeFanoutProfileConfig(value: unknown, path: string): Hwla
|
||||
agentrun: agentrunEventTypes,
|
||||
hwlab: hwlabEventTypes,
|
||||
},
|
||||
conditionalEventTypePairs,
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -36,6 +36,7 @@ test("realtime fanout keeps the YAML terminal timeout when durationMs is absent"
|
||||
},
|
||||
forbiddenRequestPaths: [],
|
||||
requiredEventTypes: { agentrun: ["terminal_status"], hwlab: ["terminal"] },
|
||||
conditionalEventTypePairs: { "tool-call": { agentrun: "tool_call", hwlab: "tool" } },
|
||||
expectedKafka: {
|
||||
topics: { stdio: "codex-stdio.raw.v1", agentrun: "agentrun.event.v1", hwlab: "hwlab.event.v1" },
|
||||
groups: { directPublish: "direct", transactionalProjector: "projector", liveSse: "live" },
|
||||
@@ -46,6 +47,91 @@ test("realtime fanout keeps the YAML terminal timeout when durationMs is absent"
|
||||
assert.equal(effectiveProfile(profile, null).terminalTimeoutMs, 480_000);
|
||||
assert.equal(effectiveProfile(profile, undefined).terminalTimeoutMs, 480_000);
|
||||
assert.equal(effectiveProfile(profile, 5_000).terminalTimeoutMs, 5_000);
|
||||
assert.deepEqual(effectiveProfile(profile, null).conditionalEventTypePairs, profile.conditionalEventTypePairs);
|
||||
});
|
||||
|
||||
test("realtime fanout terminal completion does not require absent optional tool families", () => {
|
||||
const source = nodeWebObserveRunnerRealtimeSource();
|
||||
const build = new Function(`${source}\nreturn realtimeDebugTurnCompletion;`) as () => (
|
||||
snapshot: Record<string, any>, traceId: string, profile: Record<string, any>,
|
||||
) => Record<string, any>;
|
||||
const completion = build();
|
||||
const traceId = "trc_direct_answer";
|
||||
const profile = {
|
||||
debugStreams: ["stdio", "agentrun", "hwlab"],
|
||||
requiredEventTypes: {
|
||||
agentrun: ["user_message", "assistant_message", "terminal_status"],
|
||||
hwlab: ["user", "assistant", "terminal"],
|
||||
},
|
||||
conditionalEventTypePairs: {
|
||||
"tool-call": { agentrun: "tool_call", hwlab: "tool" },
|
||||
"command-output": { agentrun: "command_output", hwlab: "status" },
|
||||
},
|
||||
};
|
||||
const snapshot = {
|
||||
records: {
|
||||
stdio: [{ traceId, eventType: "codex.stdio.stdout" }],
|
||||
agentrun: ["user_message", "backend_status", "assistant_message", "terminal_status"].map((eventType) => ({ traceId, eventType })),
|
||||
hwlab: ["user", "backend", "assistant", "terminal"].map((eventType) => ({ traceId, eventType })),
|
||||
},
|
||||
};
|
||||
|
||||
const directAnswer = completion(snapshot, traceId, profile);
|
||||
assert.equal(directAnswer.complete, true);
|
||||
assert.deepEqual(directAnswer.missingEventTypes, { agentrun: [], hwlab: [] });
|
||||
|
||||
const missingTerminal = completion({
|
||||
records: {
|
||||
...snapshot.records,
|
||||
agentrun: snapshot.records.agentrun.filter((row: Record<string, string>) => row.eventType !== "terminal_status"),
|
||||
hwlab: snapshot.records.hwlab.filter((row: Record<string, string>) => row.eventType !== "terminal"),
|
||||
},
|
||||
}, traceId, profile);
|
||||
assert.equal(missingTerminal.complete, false);
|
||||
assert.deepEqual(missingTerminal.missingEventTypes, { agentrun: ["terminal_status"], hwlab: ["terminal"] });
|
||||
assert.ok(missingTerminal.observedEventTypes.agentrun.includes("assistant_message"));
|
||||
});
|
||||
|
||||
test("realtime fanout validates optional tool events only when they occur", () => {
|
||||
const source = nodeWebObserveRunnerRealtimeSource();
|
||||
const build = new Function(`${source}\nreturn realtimeValidateConditionalEventTypePairs;`) as () => (
|
||||
input: Record<string, any>,
|
||||
) => void;
|
||||
const validate = build();
|
||||
const traceId = "trc_optional_tool";
|
||||
const base = {
|
||||
turn: 1,
|
||||
traceId,
|
||||
profile: {
|
||||
conditionalEventTypePairs: {
|
||||
"tool-call": { agentrun: "tool_call", hwlab: "tool" },
|
||||
"command-output": { agentrun: "command_output", hwlab: "status" },
|
||||
},
|
||||
},
|
||||
};
|
||||
|
||||
assert.doesNotThrow(() => validate({ ...base, debug: { records: { agentrun: [], hwlab: [] } } }));
|
||||
assert.doesNotThrow(() => validate({
|
||||
...base,
|
||||
debug: {
|
||||
records: {
|
||||
agentrun: [{ traceId, eventType: "tool_call", eventId: "evt_tool" }],
|
||||
hwlab: [{ traceId, eventType: "tool", sourceEventId: "evt_tool" }],
|
||||
},
|
||||
},
|
||||
}));
|
||||
assert.throws(
|
||||
() => validate({
|
||||
...base,
|
||||
debug: {
|
||||
records: {
|
||||
agentrun: [{ traceId, eventType: "command_output", eventId: "evt_output" }],
|
||||
hwlab: [],
|
||||
},
|
||||
},
|
||||
}),
|
||||
/conditional event pair command-output count differs/u,
|
||||
);
|
||||
});
|
||||
|
||||
test("realtime fanout retries a YAML-bounded transient health navigation", async () => {
|
||||
|
||||
@@ -747,6 +747,7 @@ function realtimeFanoutEffectiveProfile(profile, durationMs) {
|
||||
reconnectQuietMs: Number(profile.reconnectQuietMs),
|
||||
forbiddenRequestPaths: Array.isArray(profile.forbiddenRequestPaths) ? profile.forbiddenRequestPaths.map(String) : [],
|
||||
requiredEventTypes: profile.requiredEventTypes && typeof profile.requiredEventTypes === "object" ? profile.requiredEventTypes : {},
|
||||
conditionalEventTypePairs: profile.conditionalEventTypePairs && typeof profile.conditionalEventTypePairs === "object" ? sanitize(profile.conditionalEventTypePairs) : {},
|
||||
expectedProductSse: sanitize(expectedProductSse),
|
||||
expectedKafka: sanitize(expectedKafka),
|
||||
};
|
||||
@@ -1043,15 +1044,44 @@ async function realtimeWaitTurnIds(debug, key, traceId, timeoutMs) {
|
||||
}
|
||||
|
||||
async function realtimeWaitDebugTurnComplete(debug, key, traceId, profile) {
|
||||
return realtimePoll(async () => {
|
||||
const snapshot = await realtimeDebugSnapshot(debug, key);
|
||||
if (!profile.debugStreams.every((stream) => (snapshot.records[stream] || []).some((item) => item.traceId === traceId))) return null;
|
||||
const agentrunTypes = new Set((snapshot.records.agentrun || []).filter((item) => item.traceId === traceId).map((item) => item.eventType));
|
||||
const hwlabTypes = new Set((snapshot.records.hwlab || []).filter((item) => item.traceId === traceId).map((item) => item.eventType));
|
||||
if (!(profile.requiredEventTypes.agentrun || []).every((type) => agentrunTypes.has(type))) return null;
|
||||
if (!(profile.requiredEventTypes.hwlab || []).every((type) => hwlabTypes.has(type))) return null;
|
||||
return snapshot;
|
||||
}, profile.terminalTimeoutMs, "realtime debug terminal event families");
|
||||
let latest = null;
|
||||
try {
|
||||
return await realtimePoll(async () => {
|
||||
const snapshot = await realtimeDebugSnapshot(debug, key);
|
||||
latest = realtimeDebugTurnCompletion(snapshot, traceId, profile);
|
||||
return latest.complete ? snapshot : null;
|
||||
}, profile.terminalTimeoutMs, "realtime debug terminal event families");
|
||||
} catch (error) {
|
||||
const wrapped = error instanceof Error ? error : new Error(String(error));
|
||||
const missing = latest?.missingEventTypes || { agentrun: [], hwlab: [] };
|
||||
wrapped.message = "realtime debug terminal event families missing agentrun=[" + missing.agentrun.join(",") + "] hwlab=[" + missing.hwlab.join(",") + "] after " + profile.terminalTimeoutMs + "ms";
|
||||
wrapped.details = { ...(wrapped.details || {}), traceId, completion: latest, valuesRedacted: true };
|
||||
throw wrapped;
|
||||
}
|
||||
}
|
||||
|
||||
function realtimeDebugTurnCompletion(snapshot, traceId, profile) {
|
||||
const observedEventTypes = {};
|
||||
const recordCounts = {};
|
||||
for (const stream of profile.debugStreams || []) {
|
||||
const rows = (snapshot.records?.[stream] || []).filter((item) => item.traceId === traceId);
|
||||
observedEventTypes[stream] = [...new Set(rows.map((item) => item.eventType).filter(Boolean))];
|
||||
recordCounts[stream] = rows.length;
|
||||
}
|
||||
const missingEventTypes = {
|
||||
agentrun: (profile.requiredEventTypes?.agentrun || []).filter((type) => !observedEventTypes.agentrun?.includes(type)),
|
||||
hwlab: (profile.requiredEventTypes?.hwlab || []).filter((type) => !observedEventTypes.hwlab?.includes(type)),
|
||||
};
|
||||
const missingStreams = (profile.debugStreams || []).filter((stream) => !recordCounts[stream]);
|
||||
return {
|
||||
complete: missingStreams.length === 0 && missingEventTypes.agentrun.length === 0 && missingEventTypes.hwlab.length === 0,
|
||||
traceId,
|
||||
observedEventTypes,
|
||||
missingEventTypes,
|
||||
missingStreams,
|
||||
recordCounts,
|
||||
valuesRedacted: true,
|
||||
};
|
||||
}
|
||||
|
||||
async function realtimeWaitTurnStable(debug, key, traceId, subscribers, profile) {
|
||||
@@ -1241,6 +1271,7 @@ function realtimeValidateTurnEvidence(input) {
|
||||
const hwlabTypes = new Set(input.debug.records.hwlab.filter((item) => item.traceId === input.traceId).map((item) => item.eventType));
|
||||
for (const type of profile.requiredEventTypes.agentrun || []) if (!agentrunTypes.has(type)) throw new Error("turn " + input.turn + " AgentRun lacks " + type);
|
||||
for (const type of profile.requiredEventTypes.hwlab || []) if (!hwlabTypes.has(type)) throw new Error("turn " + input.turn + " HWLAB lacks " + type);
|
||||
realtimeValidateConditionalEventTypePairs(input);
|
||||
const identityRows = Object.values(input.debug.records).flat().filter((item) => item.traceId === input.traceId);
|
||||
if (!input.runId || new Set(identityRows.map((item) => item.runId)).size !== 1) throw new Error("turn " + input.turn + " runId missing or mismatched across streams");
|
||||
const commandIds = new Set(identityRows.map((item) => item.commandId).filter(Boolean));
|
||||
@@ -1267,6 +1298,30 @@ function realtimeValidateTurnEvidence(input) {
|
||||
}
|
||||
}
|
||||
|
||||
function realtimeValidateConditionalEventTypePairs(input) {
|
||||
const pairs = input.profile.conditionalEventTypePairs || {};
|
||||
const agentrunRows = (input.debug.records.agentrun || []).filter((item) => item.traceId === input.traceId);
|
||||
const hwlabRows = (input.debug.records.hwlab || []).filter((item) => item.traceId === input.traceId);
|
||||
for (const [id, pair] of Object.entries(pairs)) {
|
||||
const sourceRows = agentrunRows.filter((item) => item.eventType === pair.agentrun);
|
||||
const projectedRows = hwlabRows.filter((item) => item.eventType === pair.hwlab);
|
||||
if (sourceRows.length === 0 && projectedRows.length === 0) continue;
|
||||
if (sourceRows.length !== projectedRows.length) {
|
||||
throw new Error("turn " + input.turn + " conditional event pair " + id + " count differs AgentRun=" + sourceRows.length + " HWLAB=" + projectedRows.length);
|
||||
}
|
||||
for (const source of sourceRows) {
|
||||
if (!source.eventId || !projectedRows.some((projected) => projected.sourceEventId === source.eventId)) {
|
||||
throw new Error("turn " + input.turn + " conditional event pair " + id + " lacks HWLAB sourceEventId lineage");
|
||||
}
|
||||
}
|
||||
for (const projected of projectedRows) {
|
||||
if (!projected.sourceEventId || !sourceRows.some((source) => source.eventId === projected.sourceEventId)) {
|
||||
throw new Error("turn " + input.turn + " conditional event pair " + id + " has an orphan HWLAB event");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function realtimeExpectedStreamSessionId(stream, hwlabSessionId) {
|
||||
if (stream !== "stdio") return hwlabSessionId;
|
||||
const base = String(hwlabSessionId || "")
|
||||
|
||||
@@ -28,6 +28,14 @@ test("realtime fanout runner contract derives topics, groups, and independent ca
|
||||
projectionOutboxRelay: false,
|
||||
projectionRealtime: false,
|
||||
});
|
||||
assert.deepEqual(profiles["pure-kafka-live"]?.requiredEventTypes, {
|
||||
agentrun: ["user_message", "assistant_message", "terminal_status"],
|
||||
hwlab: ["user", "assistant", "terminal"],
|
||||
});
|
||||
assert.deepEqual(profiles["pure-kafka-live"]?.conditionalEventTypePairs, {
|
||||
"tool-call": { agentrun: "tool_call", hwlab: "tool" },
|
||||
"command-output": { agentrun: "command_output", hwlab: "status" },
|
||||
});
|
||||
});
|
||||
|
||||
test("Workbench Kafka debug replay runner contract is derived from owning YAML", () => {
|
||||
|
||||
@@ -41,8 +41,14 @@ test("validateRealtimeFanout parses the YAML profile and two independent prompts
|
||||
refreshHandoff: true,
|
||||
replayPriorTraceOnReconnect: true,
|
||||
});
|
||||
assert.ok(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.requiredEventTypes.agentrun.includes("user_message"));
|
||||
assert.ok(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.requiredEventTypes.hwlab.includes("user"));
|
||||
assert.deepEqual(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.requiredEventTypes, {
|
||||
agentrun: ["user_message", "assistant_message", "terminal_status"],
|
||||
hwlab: ["user", "assistant", "terminal"],
|
||||
});
|
||||
assert.deepEqual(spec.webProbe?.realtimeFanoutProfiles?.["pure-kafka-live"]?.conditionalEventTypePairs, {
|
||||
"tool-call": { agentrun: "tool_call", hwlab: "tool" },
|
||||
"command-output": { agentrun: "command_output", hwlab: "status" },
|
||||
});
|
||||
});
|
||||
|
||||
test("validateRealtimeFanout refuses body as a substitute for the second prompt", () => {
|
||||
|
||||
Reference in New Issue
Block a user