From b731fdfe1b6ed961a1f525992dcd4f833db641f8 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Sat, 17 Oct 2026 15:02:50 -0700 Subject: [PATCH] feat(history): normalize bounded execution evidence (#209) --- tests/test_execution_history_events.py | 346 ++++++++++++++++++++++++ utils/execution_history_events.py | 352 +++++++++++++++++++++++++ 2 files changed, 698 insertions(+) create mode 100644 tests/test_execution_history_events.py create mode 100644 utils/execution_history_events.py diff --git a/tests/test_execution_history_events.py b/tests/test_execution_history_events.py new file mode 100644 index 0000000..0aedd71 --- /dev/null +++ b/tests/test_execution_history_events.py @@ -0,0 +1,346 @@ +"""Bounded, immutable evidence captured from real host-shaped queue callbacks.""" + +from collections import namedtuple +from dataclasses import FrozenInstanceError, dataclass, replace +import json +import unittest +from uuid import UUID, uuid5 + +from utils.execution_history_events import ( + LifecycleEvent, + RunFinish, + TerminalEvidence, + classify_evidence, + encode_command, + make_finish, + make_start, +) + +SESSION = "11111111-1111-4111-8111-111111111111" +JOB = "22222222-2222-4222-8222-222222222222" +OTHER_JOB = "33333333-3333-4333-8333-333333333333" +EPOCH = "44444444-4444-4444-8444-444444444444" +Status = namedtuple("ExecutionStatus", "status_str completed messages") + + +def queue_result(item_id=0, job_id=JOB): + return ((1.0, job_id, {"secret": "graph"}, {"token": "private"}, []), item_id) + + +def message(kind, job_id=JOB, **fields): + return (kind, {"prompt_id": job_id, **fields}) + + +def finish(messages, completed=True): + return make_finish(SESSION, 0, EPOCH, Status("success", completed, messages), 500) + + +class ExecutionHistoryEventsTests(unittest.TestCase): + def test_dequeue_identity_is_deterministic_per_session_and_item(self): + start = make_start(SESSION, queue_result(), EPOCH, 100) + repeated = make_start(SESSION, queue_result(), EPOCH, 100) + second = make_start(SESSION, queue_result(1), EPOCH, 100) + self.assertEqual(start.run_id, repeated.run_id) + self.assertNotEqual(start.run_id, second.run_id) + self.assertEqual(start.run_id, str(uuid5(UUID(SESSION), "0"))) + self.assertEqual(start.queue_item_id, 0) + self.assertEqual(start.job_id, JOB) + with self.assertRaises(FrozenInstanceError): + start.job_id = OTHER_JOB + + def test_required_identity_and_observation_inputs_are_rejected(self): + invalid = [None, "bad", JOB.upper(), "", True, 1] + # The UUID fixture contains digits only; use an upper-case UUID with letters. + invalid[2] = "AAAAAAAA-AAAA-4AAA-8AAA-AAAAAAAAAAAA" + for value in invalid: + with self.subTest(value=value): + self.assertIsNone(make_start(value, queue_result(), EPOCH, 100)) + self.assertIsNone( + make_start(SESSION, queue_result(job_id=value), EPOCH, 100) + ) + self.assertIsNone(make_start(SESSION, queue_result(), value, 100)) + self.assertIsNone(make_finish(value, 0, EPOCH, None, 100)) + for value in [-1, True, 1.5, "0", 2**63]: + with self.subTest(value=value): + self.assertIsNone(make_start(SESSION, queue_result(value), EPOCH, 100)) + self.assertIsNone(make_finish(SESSION, value, EPOCH, None, 100)) + self.assertIsNone(make_start(SESSION, queue_result(), EPOCH, value)) + self.assertIsNone(make_finish(SESSION, 0, EPOCH, None, value)) + for malformed in [None, (), ((), 0), ("bad", 0), (queue_result()[0],)]: + self.assertIsNone(make_start(SESSION, malformed, EPOCH, 0)) + + def test_terminal_messages_override_aggregate_flags_and_keep_safe_fields(self): + command = finish( + [ + message("execution_start", timestamp=100), + message( + "execution_error", + timestamp=200, + node_id="9", + node_type="Sampler", + exception_type="RuntimeError", + exception_message="SECRET", + traceback=["SECRET"], + inputs={"SECRET": 1}, + ), + ] + ) + result = classify_evidence(JOB, command.evidence) + self.assertEqual((result.outcome, result.reason), ("failed", "terminal_error")) + self.assertEqual( + (result.executor_started_at, result.terminal_event_at), (100, 200) + ) + self.assertEqual( + (result.node_id, result.node_type, result.exception_type), + ("9", "Sampler", "RuntimeError"), + ) + self.assertNotIn(b"SECRET", encode_command(command)) + self.assertNotIn(b"completed", encode_command(command)) + + def test_success_interruption_and_duplicate_terminals(self): + for kind, outcome, reason in [ + ("execution_success", "succeeded", "terminal_success"), + ("execution_interrupted", "interrupted", "terminal_interrupted"), + ]: + with self.subTest(kind=kind): + event = message(kind, timestamp=200) + result = classify_evidence(JOB, finish([event, event], False).evidence) + self.assertEqual((result.outcome, result.reason), (outcome, reason)) + result = classify_evidence( + JOB, + finish( + [message("execution_success"), message("execution_interrupted")] + ).evidence, + ) + self.assertEqual( + (result.outcome, result.reason), ("unknown", "conflicting_terminal") + ) + + def test_absent_unknown_and_foreign_messages_do_not_prove_completion(self): + for messages in [ + [], + [message("executed")], + [message("execution_success", OTHER_JOB)], + ]: + with self.subTest(messages=messages): + result = classify_evidence(JOB, finish(messages).evidence) + self.assertEqual( + (result.outcome, result.reason), ("unknown", "missing_terminal") + ) + for status in [ + None, + Status("success", True, None), + Status("success", True, {}), + ]: + command = make_finish(SESSION, 0, EPOCH, status, 100) + self.assertEqual( + classify_evidence(JOB, command.evidence).reason, "invalid_evidence" + ) + + def test_malformed_evidence_is_unknown_even_with_success(self): + for malformed in [ + None, + (), + ("execution_error",), + ("execution_error", None), + ("execution_error", {}), + message("execution_error", timestamp=-1), + message("execution_error", timestamp=True), + message("execution_error", timestamp=float("nan")), + ]: + with self.subTest(malformed=malformed): + result = classify_evidence( + JOB, finish([message("execution_success"), malformed]).evidence + ) + self.assertEqual( + (result.outcome, result.reason), ("unknown", "invalid_evidence") + ) + + def test_missing_timestamps_remain_null(self): + result = classify_evidence( + JOB, + finish( + [ + message("execution_start", timestamp=None), + message("execution_success"), + ] + ).evidence, + ) + self.assertEqual(result.outcome, "succeeded") + self.assertIsNone(result.executor_started_at) + self.assertIsNone(result.terminal_event_at) + + def test_optional_oversize_and_invalid_fields_are_omitted_without_losing_failure( + self, + ): + for value in ["x" * 129, "x" * 1_000_000, {}, 12]: + with self.subTest(value_type=type(value).__name__): + command = finish( + [ + message( + "execution_error", + node_id=value, + node_type=value, + exception_type="x" * 257, + ) + ] + ) + result = classify_evidence(JOB, command.evidence) + self.assertEqual(result.outcome, "failed") + self.assertTrue(result.diagnostics_omitted) + self.assertIsNone(result.node_id) + self.assertIsNone(result.node_type) + self.assertIsNone(result.exception_type) + + def test_unicode_diagnostics_fit_utf8_budget_and_preserve_failure(self): + command = finish( + [ + message( + "execution_error", + node_id="😀" * 128, + node_type="😀" * 128, + exception_type="😀" * 256, + ) + ] + ) + encoded = encode_command(command) + self.assertLessEqual(len(encoded), 2048) + self.assertIsInstance(json.loads(encoded), dict) + result = classify_evidence(JOB, command.evidence) + self.assertEqual(result.outcome, "failed") + self.assertTrue(result.diagnostics_omitted) + + def test_message_limit_never_trusts_partial_success(self): + command = finish([message("execution_success")] * 1025) + result = classify_evidence(JOB, command.evidence) + self.assertEqual((result.outcome, result.reason), ("unknown", "evidence_limit")) + self.assertLessEqual(len(encode_command(command)), 2048) + self.assertEqual( + classify_evidence( + JOB, finish([message("execution_success")] * 1024).evidence + ).outcome, + "succeeded", + ) + + def test_required_event_budget_exhaustion_is_not_truncated_success(self): + messages = [message("execution_success", timestamp=i) for i in range(100)] + command = finish(messages) + self.assertEqual( + classify_evidence(JOB, command.evidence).reason, "evidence_limit" + ) + self.assertLessEqual(len(encode_command(command)), 2048) + + def test_classification_uses_stored_job_id_after_capture(self): + command = finish( + [message("execution_error", OTHER_JOB), message("execution_success", JOB)] + ) + self.assertEqual(classify_evidence(JOB, command.evidence).outcome, "succeeded") + self.assertEqual( + classify_evidence(OTHER_JOB, command.evidence).outcome, "failed" + ) + + def test_encode_rejects_forged_invalid_or_unbounded_commands(self): + start = make_start(SESSION, queue_result(), EPOCH, 100) + command = finish([message("execution_success")]) + bad_event = LifecycleEvent("execution_success", JOB, -1, None, None, None) + for invalid in [ + None, + {}, + replace(start, run_id="bad"), + replace(start, job_id="x" * 129), + replace(start, queue_item_id=True), + replace(command, observed_callback_at=-1), + replace( + command, evidence=TerminalEvidence((bad_event,), False, False, False) + ), + replace(command, evidence=TerminalEvidence((), 1, False, False)), + replace( + command, + evidence=TerminalEvidence( + command.evidence.events * 100, False, False, False + ), + ), + ]: + with self.subTest(invalid=invalid): + with self.assertRaises(ValueError): + encode_command(invalid) + self.assertLessEqual(len(encode_command(start)), 2048) + self.assertIsInstance(command, RunFinish) + + def test_encoder_rejects_unknown_fields_and_nonstring_event_kinds(self): + @dataclass(frozen=True) + class ExtendedFinish(RunFinish): + secret: str = "credentials" + + command = finish([message("execution_success")]) + extended = ExtendedFinish(SESSION, 0, EPOCH, 100, command.evidence) + invalid_kind = LifecycleEvent([], JOB, None, None, None, None) + for invalid in [ + extended, + replace( + command, evidence=TerminalEvidence((invalid_kind,), False, False, False) + ), + ]: + with self.subTest(invalid=invalid): + with self.assertRaises(ValueError): + encode_command(invalid) + + def test_malformed_foreign_timestamp_cannot_discredit_matching_success(self): + command = finish( + [ + message("execution_error", OTHER_JOB, timestamp=-1), + message("execution_success", JOB, timestamp=300), + ] + ) + matching = classify_evidence(JOB, command.evidence) + self.assertEqual( + (matching.outcome, matching.reason), ("succeeded", "terminal_success") + ) + self.assertEqual( + classify_evidence(OTHER_JOB, command.evidence).reason, "invalid_evidence" + ) + self.assertLessEqual(len(encode_command(command)), 2048) + + def test_malformed_foreign_envelope_is_scoped_when_identity_is_usable(self): + command = finish( + [ + ("execution_error", {"prompt_id": OTHER_JOB}, "unexpected"), + message("execution_success", JOB), + ] + ) + self.assertEqual(classify_evidence(JOB, command.evidence).outcome, "succeeded") + self.assertEqual( + classify_evidence(OTHER_JOB, command.evidence).reason, "invalid_evidence" + ) + + def test_diagnostic_budget_reduction_retains_job_scoped_invalidity(self): + command = finish( + [ + message("execution_error", JOB, timestamp=-1), + message( + "execution_success", + JOB, + node_id="😀" * 128, + node_type="😀" * 128, + exception_type="😀" * 256, + ), + ] + ) + result = classify_evidence(JOB, command.evidence) + self.assertEqual( + (result.outcome, result.reason), ("unknown", "invalid_evidence") + ) + self.assertLessEqual(len(encode_command(command)), 2048) + + def test_job_scoped_invalidity_budget_overflow_is_evidence_limit(self): + command = finish( + [ + message("execution_error", str(UUID(int=i)), timestamp=-1) + for i in range(100) + ] + + [message("execution_success")] + ) + self.assertEqual( + classify_evidence(JOB, command.evidence).reason, "evidence_limit" + ) + self.assertLessEqual(len(encode_command(command)), 2048) diff --git a/utils/execution_history_events.py b/utils/execution_history_events.py new file mode 100644 index 0000000..5be4f6b --- /dev/null +++ b/utils/execution_history_events.py @@ -0,0 +1,352 @@ +"""Pure, bounded execution-lifecycle records for the background ledger writer.""" + +from dataclasses import asdict, dataclass, replace +import json +from uuid import UUID, uuid5 + +MAX_COMMAND_BYTES = 2048 +MAX_MESSAGES = 1024 +MAX_INTEGER = 2**63 - 1 +TERMINAL_KINDS = { + "execution_success": ("succeeded", "terminal_success"), + "execution_error": ("failed", "terminal_error"), + "execution_interrupted": ("interrupted", "terminal_interrupted"), +} +EVENT_KINDS = frozenset((*TERMINAL_KINDS, "execution_start")) + + +@dataclass(frozen=True) +class RunStart: + run_id: str + session_id: str + queue_item_id: int + job_id: str + storage_epoch: str + observed_dequeue_at: int + + +@dataclass(frozen=True) +class LifecycleEvent: + kind: str + job_id: str + timestamp: int | None + node_id: str | None + node_type: str | None + exception_type: str | None + + +@dataclass(frozen=True) +class TerminalEvidence: + events: tuple[LifecycleEvent, ...] + invalid: bool + limited: bool + diagnostics_omitted: bool + invalid_job_ids: tuple[str, ...] = () + + +@dataclass(frozen=True) +class RunFinish: + session_id: str + queue_item_id: int + storage_epoch: str + observed_callback_at: int + evidence: TerminalEvidence + + +@dataclass(frozen=True) +class Completion: + outcome: str + reason: str + executor_started_at: int | None + terminal_event_at: int | None + node_id: str | None + node_type: str | None + exception_type: str | None + diagnostics_omitted: bool + + +RunCommand = RunStart | RunFinish + + +def _valid_integer(value: object) -> bool: + return type(value) is int and 0 <= value <= MAX_INTEGER + + +def _valid_uuid(value: object) -> bool: + if not isinstance(value, str) or len(value) != 36: + return False + try: + return str(UUID(value)) == value + except ValueError: + return False + + +def _valid_identity(session_id: object, item_id: object, epoch: object) -> bool: + return _valid_uuid(session_id) and _valid_integer(item_id) and _valid_uuid(epoch) + + +def make_start( + session_id: str, queue_result: object, storage_epoch: str, observed_at: int +) -> RunStart | None: + """Extract only the public job ID and queue item ID from a dequeue result.""" + if not isinstance(queue_result, (tuple, list)) or len(queue_result) != 2: + return None + queue_item, item_id = queue_result + if not isinstance(queue_item, (tuple, list)) or len(queue_item) < 2: + return None + job_id = queue_item[1] + if not ( + _valid_identity(session_id, item_id, storage_epoch) + and _valid_uuid(job_id) + and _valid_integer(observed_at) + ): + return None + return RunStart( + str(uuid5(UUID(session_id), str(item_id))), + session_id, + item_id, + job_id, + storage_epoch, + observed_at, + ) + + +def _optional_diagnostic(value: object, limit: int) -> tuple[str | None, bool]: + if value is None: + return None, False + if isinstance(value, str) and len(value) <= limit: + # Lone surrogates cannot be encoded as UTF-8 allowed data. + try: + value.encode("utf-8") + except UnicodeEncodeError: + return None, True + return value, False + return None, True + + +def make_finish( + session_id: str, + queue_item_id: int, + storage_epoch: str, + status: object, + observed_at: int, +) -> RunFinish | None: + """Capture allowed lifecycle fields without copying the host envelope.""" + if not ( + _valid_identity(session_id, queue_item_id, storage_epoch) + and _valid_integer(observed_at) + ): + return None + messages = getattr(status, "messages", None) + events = {} + invalid_job_ids = {} + invalid = not isinstance(messages, (tuple, list)) + limited = False + omitted = False + if not invalid: + limited = len(messages) > MAX_MESSAGES + for index in range(min(len(messages), MAX_MESSAGES)): + message = messages[index] + usable_job_id = None + if ( + isinstance(message, (tuple, list)) + and len(message) >= 2 + and isinstance(message[1], dict) + and _valid_uuid(message[1].get("prompt_id")) + ): + usable_job_id = message[1]["prompt_id"] + malformed = not isinstance(message, (tuple, list)) or len(message) != 2 + if not malformed: + kind, data = message + malformed = not isinstance(kind, str) + if not malformed and kind not in EVENT_KINDS: + continue + malformed |= usable_job_id is None + if not malformed: + timestamp = data.get("timestamp") + malformed = timestamp is not None and not _valid_integer(timestamp) + if malformed: + if usable_job_id is None: + invalid = True + else: + invalid_job_ids[usable_job_id] = None + continue + diagnostics = [] + for field, limit in ( + ("node_id", 128), + ("node_type", 128), + ("exception_type", 256), + ): + value, dropped = _optional_diagnostic(data.get(field), limit) + diagnostics.append(value) + omitted |= dropped + event = LifecycleEvent(kind, data["prompt_id"], timestamp, *diagnostics) + events[event] = None + command = RunFinish( + session_id, + queue_item_id, + storage_epoch, + observed_at, + TerminalEvidence( + tuple(events), invalid, limited, omitted, tuple(invalid_job_ids) + ), + ) + if len(_serialize(command)) <= MAX_COMMAND_BYTES: + return command + # Diagnostics yield to all candidate identities, kinds and event times. + stripped = tuple( + dict.fromkeys( + replace(event, node_id=None, node_type=None, exception_type=None) + for event in command.evidence.events + ) + ) + had_diagnostics = any( + event.node_id is not None + or event.node_type is not None + or event.exception_type is not None + for event in command.evidence.events + ) + command = replace( + command, + evidence=TerminalEvidence( + stripped, + invalid, + limited, + omitted or had_diagnostics, + tuple(invalid_job_ids), + ), + ) + if len(_serialize(command)) > MAX_COMMAND_BYTES: + # Required evidence could not fit: no partial terminal may prove success. + command = replace( + command, + evidence=replace( + command.evidence, events=(), limited=True, invalid_job_ids=() + ), + ) + return command + + +def classify_evidence(job_id: str, evidence: TerminalEvidence) -> Completion: + """Classify only candidates matching the job ID persisted at dequeue.""" + events = tuple(event for event in evidence.events if event.job_id == job_id) + start_times = [ + event.timestamp + for event in events + if event.kind == "execution_start" and event.timestamp is not None + ] + started_at = min(start_times, default=None) + terminals = tuple(event for event in events if event.kind in TERMINAL_KINDS) + kinds = {event.kind for event in terminals} + reason = None + if evidence.limited: + reason = "evidence_limit" + elif evidence.invalid or job_id in evidence.invalid_job_ids: + reason = "invalid_evidence" + elif len(kinds) > 1: + reason = "conflicting_terminal" + elif not kinds: + reason = "missing_terminal" + if reason is not None: + return Completion( + "unknown", + reason, + started_at, + None, + None, + None, + None, + evidence.diagnostics_omitted, + ) + outcome, reason = TERMINAL_KINDS[terminals[0].kind] + terminal_times = [ + event.timestamp for event in terminals if event.timestamp is not None + ] + return Completion( + outcome, + reason, + started_at, + max(terminal_times, default=None), + next((event.node_id for event in terminals if event.node_id is not None), None), + next( + (event.node_type for event in terminals if event.node_type is not None), + None, + ), + next( + ( + event.exception_type + for event in terminals + if event.exception_type is not None + ), + None, + ), + evidence.diagnostics_omitted, + ) + + +def _valid_evidence(evidence: object) -> bool: + if type(evidence) is not TerminalEvidence or not isinstance(evidence.events, tuple): + return False + if len(evidence.events) > MAX_MESSAGES or any( + type(flag) is not bool + for flag in (evidence.invalid, evidence.limited, evidence.diagnostics_omitted) + ): + return False + if ( + not isinstance(evidence.invalid_job_ids, tuple) + or len(evidence.invalid_job_ids) > MAX_MESSAGES + or not all(_valid_uuid(job_id) for job_id in evidence.invalid_job_ids) + ): + return False + for event in evidence.events: + if type(event) is not LifecycleEvent: + return False + if ( + not isinstance(event.kind, str) + or event.kind not in EVENT_KINDS + or not _valid_uuid(event.job_id) + or (event.timestamp is not None and not _valid_integer(event.timestamp)) + ): + return False + for value, limit in ( + (event.node_id, 128), + (event.node_type, 128), + (event.exception_type, 256), + ): + if _optional_diagnostic(value, limit)[1]: + return False + return True + + +def _serialize(command: RunCommand) -> bytes: + return json.dumps( + asdict(command), ensure_ascii=False, allow_nan=False, separators=(",", ":") + ).encode("utf-8") + + +def encode_command(command: RunCommand) -> bytes: + """Validate and encode an immutable command within the transport budget.""" + if type(command) not in (RunStart, RunFinish): + raise ValueError("Invalid execution history command") + valid = _valid_identity( + command.session_id, command.queue_item_id, command.storage_epoch + ) + if isinstance(command, RunStart): + valid = ( + valid + and _valid_uuid(command.run_id) + and _valid_uuid(command.job_id) + and _valid_integer(command.observed_dequeue_at) + ) + else: + valid = ( + valid + and _valid_integer(command.observed_callback_at) + and _valid_evidence(command.evidence) + ) + if not valid: + raise ValueError("Invalid execution history command") + encoded = _serialize(command) + if len(encoded) > MAX_COMMAND_BYTES: + raise ValueError("Execution history command exceeds byte limit") + return encoded