feat(history): normalize bounded execution evidence (#209)
This commit is contained in:
Vito Sansevero
@@ -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)
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user