fix(anchor): harden normalizer payload shaping

This commit is contained in:
Frank Song committed 2026-06-12 13:57:54 +08:00
1 parent 36123628d5
commit 2abff38ee1
2 files changed
+79 -8

No files matched your search

+20 -7
View File
@@ -143,6 +143,8 @@
if(Array.isArray(value)){
return value.map((item)=>_sanitizePayload(item,depth+1)).filter((item)=>item!==undefined);
}
const proto=Object.getPrototypeOf(value);
if(proto!==null&&Object.prototype.toString.call(value)!=='[object Object]') return '[Object]';
const out={};
Object.keys(value).sort().forEach((key)=>{
const safe=_sanitizePayload(value[key],depth+1);
@@ -190,6 +192,7 @@
'source_event_type',
'sourceType',
'source_type',
'event_type',
'type',
'event',
'event_id',
@@ -266,7 +269,17 @@
function normalizeAssistantTurnAnchorSourceEvent(input, context){
const event=(input&&typeof input==='object')?input:{};
const ctx=(context&&typeof context==='object')?context:{};
const payload=_sanitizePayload(_sourceEventPayload(event))||{};
const sanitizedPayload=_sanitizePayload(_sourceEventPayload(event));
const rawPayload=(sanitizedPayload&&typeof sanitizedPayload==='object'&&!Array.isArray(sanitizedPayload))?sanitizedPayload:{};
const {
session_id:_payloadSessionId,
turn_id:_payloadTurnId,
run_id:_payloadRunId,
stream_id:_payloadStreamId,
event_id:_payloadEventId,
seq:_payloadSeq,
...payload
}=rawPayload;
const sourceType=_sourceEventType(event,payload);
const meta=classifyAssistantTurnAnchorSourceEvent(sourceType);
const classification=meta.classification;
@@ -278,17 +291,17 @@
dedupe_key:'',
});
}
const eventId=_cleanString(event.event_id||event.lastEventId||event.last_event_id||payload.event_id);
const eventId=_cleanString(event.event_id||event.lastEventId||event.last_event_id||rawPayload.event_id);
const seq=_coerceSeq(
event.seq!==undefined?event.seq:
payload.seq!==undefined?payload.seq:
rawPayload.seq!==undefined?rawPayload.seq:
ctx.seq!==undefined?ctx.seq:
_eventIdSeq(eventId)
);
const runId=_cleanString(event.run_id||payload.run_id||ctx.run_id)||_eventIdRunId(eventId)||null;
const sessionId=_cleanString(event.session_id||payload.session_id||ctx.session_id);
const turnId=_cleanString(event.turn_id||payload.turn_id||ctx.turn_id);
const streamId=_cleanString(event.stream_id||payload.stream_id||ctx.stream_id)||null;
const runId=_cleanString(event.run_id||rawPayload.run_id||ctx.run_id)||_eventIdRunId(eventId)||null;
const sessionId=_cleanString(event.session_id||rawPayload.session_id||ctx.session_id);
const turnId=_cleanString(event.turn_id||rawPayload.turn_id||ctx.turn_id);
const streamId=_cleanString(event.stream_id||rawPayload.stream_id||ctx.stream_id)||null;
const localId=_localIdForSourceEvent(sourceType, {...ctx,seq}, payload);
const anchorEvent={
event_id:eventId||null,
@@ -89,6 +89,30 @@ const directParsedPayload = api.normalizeAssistantTurnAnchorSourceEvent({{
event_id:'run-1:14',
seq:14,
}}, context);
const eventTypePayload = api.normalizeAssistantTurnAnchorSourceEvent({{
event_type:'tool',
event_id:'run-1:15',
tool_call_id:'tool-1',
someField:'x',
}}, context);
const nonPlainPayload = api.normalizeAssistantTurnAnchorSourceEvent({{
type:'token',
payload:{{text:'date payload', meta:new Date('2026-06-11T00:00:00Z')}},
event_id:'run-1:16',
}}, context);
const promotedIdentityPayload = api.normalizeAssistantTurnAnchorSourceEvent({{
type:'token',
payload:{{
text:'identity fields should be promoted only',
session_id:'payload-session',
turn_id:'payload-turn',
run_id:'payload-run',
stream_id:'payload-stream',
event_id:'payload-event',
seq:99,
}},
event_id:'run-1:17',
}}, context);
console.log(JSON.stringify({{
version: api.version,
liveToken,
@@ -101,6 +125,9 @@ console.log(JSON.stringify({{
deduped,
invalidPayload,
directParsedPayload,
eventTypePayload,
nonPlainPayload,
promotedIdentityPayload,
}}));
"""
result = subprocess.run([NODE, "-e", script], text=True, capture_output=True, check=False)
@@ -124,7 +151,7 @@ def test_normalizer_maps_live_and_replay_to_same_anchor_event_identity():
assert live["anchor_event"]["run_id"] == "run-1"
assert live["anchor_event"]["stream_id"] == "stream-1"
assert live["anchor_event"]["seq"] == 7
assert live["anchor_event"]["payload"] == {"session_id": "sid-1", "text": "hello"}
assert live["anchor_event"]["payload"] == {"text": "hello"}
def test_normalizer_keeps_artifact_side_effect_metadata_and_transport_distinct():
@@ -168,6 +195,37 @@ def test_normalizer_excludes_unknown_sources_and_sanitizes_non_json_data():
assert data["directParsedPayload"]["anchor_event"]["seq"] == 14
def test_normalizer_strips_discriminators_and_promoted_identity_fields_from_payload():
data = _normalizer_snapshot()
event_type_payload = data["eventTypePayload"]
assert event_type_payload["anchor_event"]["source_event_type"] == "tool"
assert event_type_payload["anchor_event"]["payload"] == {
"someField": "x",
"tool_call_id": "tool-1",
}
promoted = data["promotedIdentityPayload"]
assert promoted["anchor_event"]["event_id"] == "run-1:17"
assert promoted["anchor_event"]["session_id"] == "payload-session"
assert promoted["anchor_event"]["turn_id"] == "payload-turn"
assert promoted["anchor_event"]["run_id"] == "payload-run"
assert promoted["anchor_event"]["stream_id"] == "payload-stream"
assert promoted["anchor_event"]["seq"] == 99
assert promoted["anchor_event"]["payload"] == {
"text": "identity fields should be promoted only"
}
def test_normalizer_marks_non_plain_payload_objects_explicitly():
data = _normalizer_snapshot()
assert data["nonPlainPayload"]["anchor_event"]["payload"] == {
"meta": "[Object]",
"text": "date payload",
}
def test_batch_normalizer_dedupes_live_plus_replay_by_event_envelope():
data = _normalizer_snapshot()
deduped = data["deduped"]