Vito
763e6b37b1
* feat(history): normalize bounded execution evidence (#209) * fix(history): bound lifecycle kind checks before hashing (#209) * feat(history): persist execution attempts with restore fencing (#209) * test(api): bootstrap standalone backup restore fixtures (#209) * feat(history): isolate bounded asynchronous ledger writes (#209) * feat(history): observe actual queue attempts without altering execution (#209) * fix(database): honor package-scoped configuration (#211) * test(history): exercise real cached and headless executions (#209) * fix(history): reject unsupported uniqueness and isolate package fixtures (#209) * test(history): support Python 3.10 package fixture cleanup (#209)
380 lines
15 KiB
Python
380 lines
15 KiB
Python
"""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)
|
|
|
|
def test_oversized_unknown_kind_is_ignored_without_hashing(self):
|
|
class OversizedKind(str):
|
|
def __hash__(self):
|
|
raise AssertionError("Oversized event kind must not be hashed")
|
|
|
|
command = finish(
|
|
[
|
|
message(OversizedKind("x" * 1_000_000)),
|
|
message("execution_success", timestamp=300),
|
|
]
|
|
)
|
|
result = classify_evidence(JOB, command.evidence)
|
|
self.assertEqual(
|
|
(result.outcome, result.reason), ("succeeded", "terminal_success")
|
|
)
|
|
self.assertLessEqual(len(encode_command(command)), 2048)
|
|
|
|
def test_encoder_rejects_oversized_kind_without_hashing(self):
|
|
class OversizedKind(str):
|
|
def __hash__(self):
|
|
raise AssertionError("Oversized event kind must not be hashed")
|
|
|
|
command = finish([message("execution_success")])
|
|
event = LifecycleEvent(
|
|
OversizedKind("x" * 1_000_000), JOB, None, None, None, None
|
|
)
|
|
with self.assertRaises(ValueError):
|
|
encode_command(
|
|
replace(
|
|
command, evidence=TerminalEvidence((event,), False, False, False)
|
|
)
|
|
)
|