from __future__ import annotations import base64 import hashlib import json import tempfile import unittest from datetime import date, datetime, timedelta, timezone from typing import List, Optional, Union from pathlib import Path from agent_integration.client import OpenAgentAPIError, OpenAgentEvent, OpenAgentTransportError from arr_ingestion.contracts import ArtifactRef, XLSX_MIME from arr_processing.contracts import ( PROCESSING_REQUEST_VERSION, ProcessingRequest, ProcessingResult, ) from arr_processing.config import ResultVerificationConfig from arr_processing.errors import ProcessingError, ProcessingTransportError from arr_processing.ledger import InMemoryProcessingLedger from arr_processing.open_agent import OpenAgentProcessingTransport from arr_processing.registration import ProcessingOutputRegistrar from arr_processing.runner import ProcessingRunner, RemoteRunSnapshot from arr_processing.signatures import SignedResultCodec from arr_processing.source_message import ( PROGRAM_INPUT_VERSION, OssProcessingMessageBuilder, ) from arr_storage.aliyun_oss_v2 import AliyunOssConfig from arr_storage import FilesystemObjectBackend, ManagedObjectStore NOW = datetime(2026, 7, 28, 8, 0, tzinfo=timezone.utc) SECRET = b"synthetic-processing-signature-key-32-bytes-minimum" def request(job_id: str = "job-processing-001", attempt_no: int = 1) -> ProcessingRequest: return ProcessingRequest( job_id=job_id, attempt_no=attempt_no, source_file_id="arr_file_source_001", processor_version="3.0.0", rule_set_sha256="a" * 64, submission_grant="G" * 43, ) def artifact(role: str, filename: str, content: bytes, mime_type: str) -> dict[str, object]: return { "file_handle": "file_" + role, "original_filename": filename, "sha256": hashlib.sha256(content).hexdigest(), "byte_size": len(content), "mime_type": mime_type, } def result_payload( *, job_id: str = "job-processing-001", attempt_no: int = 1, remote_run_id: str = "run-processing-001", delivery_id: str = "delivery-processing-001", status: str = "success", ) -> dict[str, object]: success = status == "success" return { "contract_version": "arr-opera-daily-result-1", "delivery_id": delivery_id, "job_id": job_id, "attempt_no": attempt_no, "remote_run_id": remote_run_id, "status": status, "business_date": "2026-07-28" if success else None, "processor_version": "3.0.0", "rule_set_sha256": "a" * 64, "result_schema_version": "3.0", "artifacts": { "daily_report": ( artifact("daily_report", "7.28.xlsx", b"daily", XLSX_MIME) if success else None ), "result_json": artifact( "result_json", "result.json", b"result", "application/json" ), "structured_result_json": artifact( "structured_result_json", "structured-result.json", b"structured", "application/json", ), "exception_report": ( None if success else artifact( "exception_report", "exception.xlsx", b"exception", XLSX_MIME, ) ), }, } def codec(now: datetime = NOW) -> SignedResultCodec: return SignedResultCodec( {"callback-key-1": SECRET}, signing_key_id="callback-key-1", clock=lambda: now, ) TransportStep = Union[RemoteRunSnapshot, ProcessingTransportError] class FakeProcessingTransport: def __init__( self, *, submit: Optional[List[TransportStep]] = None, get: Optional[List[TransportStep]] = None, cancel: Optional[List[TransportStep]] = None, ) -> None: self.submit_steps = list(submit or []) self.get_steps = list(get or []) self.cancel_steps = list(cancel or []) self.submit_calls: list[tuple[ProcessingRequest, str]] = [] self.get_calls: list[tuple[ProcessingRequest, str]] = [] self.cancel_calls: list[tuple[ProcessingRequest, str]] = [] @staticmethod def _next(steps: List[TransportStep]) -> RemoteRunSnapshot: if not steps: raise AssertionError("fake transport has no configured step") value = steps.pop(0) if isinstance(value, ProcessingTransportError): raise value return value def submit(self, value: ProcessingRequest, idempotency_key: str) -> RemoteRunSnapshot: self.submit_calls.append((value, idempotency_key)) return self._next(self.submit_steps) def get(self, value: ProcessingRequest, remote_run_id: str) -> RemoteRunSnapshot: self.get_calls.append((value, remote_run_id)) return self._next(self.get_steps) def cancel(self, value: ProcessingRequest, remote_run_id: str) -> RemoteRunSnapshot: self.cancel_calls.append((value, remote_run_id)) return self._next(self.cancel_steps) class ResultVerificationConfigTests(unittest.TestCase): def test_callback_secret_is_separate_base64_environment_configuration(self): encoded = base64.b64encode(SECRET).decode("ascii") config = ResultVerificationConfig.from_environment( { "ARR_AGENT_RESULT_HMAC_KEY_ID": "callback-key-1", "ARR_AGENT_RESULT_HMAC_KEY_B64": encoded, "ARR_AGENT_RESULT_MAX_AGE_SECONDS": "300", "DEERFLOW_OPEN_API_KEY": "must-not-be-used-as-callback-secret", } ) self.assertEqual(config.key_id, "callback-key-1") self.assertEqual(config.secret, SECRET) self.assertEqual(config.max_age_seconds, 300) self.assertIsInstance(config.verifier(), SignedResultCodec) with self.assertRaises(ProcessingError): ResultVerificationConfig.from_environment( { "ARR_AGENT_RESULT_HMAC_KEY_ID": "callback-key-1", "ARR_AGENT_RESULT_HMAC_KEY_B64": "not-base64", } ) class ProcessingContractTests(unittest.TestCase): def test_external_schemas_are_strict_and_path_free(self): project_root = Path(__file__).resolve().parents[1] program_schema = json.loads( (project_root / "prompts" / "arr_opera_daily_program_input.schema.json").read_text( encoding="utf-8" ) ) request_schema = json.loads( (project_root / "prompts" / "arr_opera_daily_processing_request.schema.json").read_text( encoding="utf-8" ) ) result_path = project_root / "prompts" / "arr_opera_daily_processing_result.schema.json" result_schema = json.loads(result_path.read_text(encoding="utf-8")) self.assertFalse(request_schema["additionalProperties"]) self.assertFalse(result_schema["additionalProperties"]) self.assertFalse(program_schema["additionalProperties"]) self.assertEqual( program_schema["properties"]["contract_version"]["const"], "arr-opera-daily-program-input-3", ) attachment_oss = program_schema["$defs"]["attachment"]["properties"]["oss"] self.assertIn("url", attachment_oss["required"]) self.assertEqual(attachment_oss["properties"]["url"]["format"], "uri") self.assertEqual( program_schema["properties"]["attachment_fetch_policy"] ["properties"]["tool_name"]["const"], "fetch_oss_file", ) self.assertEqual( request_schema["properties"]["contract_version"]["const"], "arr-opera-daily-request-1", ) self.assertEqual( result_schema["properties"]["signed_envelope_version"]["const"], "arr-processing-signed-1", ) rendered = result_path.read_text(encoding="utf-8") for forbidden in ("local_path", "object_key", "signed_url", "access_key"): self.assertNotIn(forbidden, rendered.lower()) def test_request_message_contains_only_opaque_contract_fields(self): value = request() payload = json.loads(value.message()) self.assertEqual(payload, value.to_dict()) self.assertEqual(payload["contract_version"], PROCESSING_REQUEST_VERSION) rendered = value.message() for forbidden in ("object_key", "local_path", "http://", "https://", "AccessKey", " ArtifactRef: self_outer.assertEqual(job_id, value.job_id) self_outer.assertEqual(attempt_no, value.attempt_no) return source self_outer = self builder = OssProcessingMessageBuilder( AliyunOssConfig( "cn-guangzhou", "one-feel-bucket", "https://oss-cn-guangzhou.aliyuncs.com", ), Sources(), ) rendered = builder.message(value) payload = json.loads(rendered) self.assertEqual(payload["contract_version"], PROGRAM_INPUT_VERSION) self.assertEqual(payload["source_file_id"], value.source_file_id) self.assertEqual(payload["oss_inline_images"], []) self.assertEqual(payload["attachment_fetch_policy"]["tool_name"], "fetch_oss_file") self.assertEqual( payload["result_submission"]["tool_name"], "arr_submit_processing_result", ) self.assertEqual( payload["result_submission"]["submission_grant"], value.submission_grant, ) self.assertNotIn(str(value.submission_grant), value.message()) self.assertNotIn(str(value.submission_grant), repr(value)) attachment = payload["oss_attachments"][0] self.assertEqual(attachment["id"], value.source_file_id) self.assertEqual(attachment["name"], "source.xml") self.assertEqual(attachment["sha256"], "b" * 64) self.assertEqual(attachment["size"], 321) self.assertEqual(attachment["oss"]["bucket"], "one-feel-bucket") self.assertEqual( attachment["oss"]["endpoint"], "oss-cn-guangzhou.aliyuncs.com", ) self.assertEqual(attachment["oss"]["object_key"], source.object_key) self.assertEqual( attachment["oss"]["url"], "https://one-feel-bucket.oss-cn-guangzhou.aliyuncs.com/" "arr/jobs/arrjob-source%3Amessage-001/attempts/0001/" "committed/source_xml/source.xml", ) self.assertNotIn("?", attachment["oss"]["url"]) for forbidden in ("access_key", "secret", "signed_url", "local_path"): self.assertNotIn(forbidden, rendered.lower()) def test_oss_program_message_rejects_mismatched_source_job(self): value = request(job_id="arrjob-source-message-001") source = ArtifactRef( role="source_xml", file_kind="opera_xml", object_key=( "arr/jobs/arrjob-other/attempts/0001/committed/source_xml/source.xml" ), original_filename="source.xml", sha256="b" * 64, byte_size=321, mime_type="application/xml", ) class Sources: def source_for_attempt(self, _job_id: str, _attempt_no: int) -> ArtifactRef: return source with self.assertRaisesRegex(ProcessingError, "source object"): OssProcessingMessageBuilder( AliyunOssConfig("cn-guangzhou", "one-feel-bucket"), Sources(), ).message(value) def test_oss_program_message_requires_ephemeral_submission_grant(self): value = ProcessingRequest( job_id="arrjob-source-message-001", attempt_no=1, source_file_id="arr_file_source_001", processor_version="3.0.0", rule_set_sha256="a" * 64, ) source = ArtifactRef( role="source_xml", file_kind="opera_xml", object_key=( "arr/jobs/arrjob-source-message-001/attempts/0001/" "committed/source_xml/source.xml" ), original_filename="source.xml", sha256="b" * 64, byte_size=321, mime_type="application/xml", ) class Sources: def source_for_attempt(self, _job_id: str, _attempt_no: int) -> ArtifactRef: return source with self.assertRaises(ProcessingError) as raised: OssProcessingMessageBuilder( AliyunOssConfig("cn-guangzhou", "one-feel-bucket"), Sources(), ).message(value) self.assertEqual( raised.exception.code, "PROCESSING_SUBMISSION_GRANT_MISSING", ) def test_result_success_and_business_failure_shapes_are_strict(self): success = ProcessingResult.from_dict(result_payload()) failure = ProcessingResult.from_dict(result_payload(status="failed")) self.assertEqual(success.business_date, date(2026, 7, 28)) self.assertIsNotNone(success.artifacts["daily_report"]) self.assertIsNone(success.artifacts["exception_report"]) self.assertIsNone(failure.artifacts["daily_report"]) self.assertIsNotNone(failure.artifacts["exception_report"]) def test_bad_artifact_metadata_and_local_paths_are_rejected(self): payload = result_payload() daily = payload["artifacts"]["daily_report"] # type: ignore[index] self.assertIsInstance(daily, dict) daily["local_path"] = "/private/output.xlsx" # type: ignore[index] with self.assertRaises(ProcessingError): ProcessingResult.from_dict(payload) payload = result_payload() payload["artifacts"]["daily_report"]["mime_type"] = "text/plain" # type: ignore[index] with self.assertRaises(ProcessingError): ProcessingResult.from_dict(payload) payload = result_payload() payload["artifacts"]["result_json"]["sha256"] = "bad" # type: ignore[index] with self.assertRaises(ProcessingError): ProcessingResult.from_dict(payload) class SignedResultTests(unittest.TestCase): def test_sign_verify_tamper_and_expiry(self): result = ProcessingResult.from_dict(result_payload()) signed = codec().sign(result, issued_at=NOW, nonce="nonce-001") verified = codec().verify(signed) self.assertEqual(verified.result, result) self.assertEqual(verified.key_id, "callback-key-1") self.assertNotIn(SECRET.decode("ascii"), signed.decode("utf-8")) tampered = signed.replace(b"7.28.xlsx", b"7.29.xlsx") with self.assertRaises(ProcessingError) as tamper_error: codec().verify(tampered) self.assertEqual(tamper_error.exception.code, "PROCESSING_SIGNATURE_INVALID") expired_codec = codec(NOW + timedelta(seconds=601)) with self.assertRaises(ProcessingError) as expired: expired_codec.verify(signed) self.assertEqual(expired.exception.code, "PROCESSING_SIGNATURE_EXPIRED") def test_duplicate_json_keys_and_unknown_key_are_rejected(self): duplicate = b'{"signed_envelope_version":"arr-processing-signed-1","signed_envelope_version":"arr-processing-signed-1"}' with self.assertRaises(ProcessingError): codec().verify(duplicate) verify_only = SignedResultCodec( {"different-key": b"another-synthetic-signature-secret-32-bytes"}, clock=lambda: NOW, ) signed = codec().sign(ProcessingResult.from_dict(result_payload()), issued_at=NOW) with self.assertRaises(ProcessingError): verify_only.verify(signed) class ProcessingRunnerTests(unittest.TestCase): def test_retryable_429_and_timeout_reuse_one_idempotency_key(self): transport = FakeProcessingTransport( submit=[ ProcessingTransportError("PROCESSING_REMOTE_RATE_LIMITED", retryable=True), ProcessingTransportError("PROCESSING_REMOTE_UNAVAILABLE", retryable=True), RemoteRunSnapshot("run-processing-001", "running"), ] ) sleeps: list[float] = [] ledger = InMemoryProcessingLedger() runner = ProcessingRunner( transport, ledger, codec(), retry_delay_seconds=0.1, sleep=sleeps.append, ) record = runner.start(request()) self.assertEqual(record.remote_run_id, "run-processing-001") self.assertEqual(len(transport.submit_calls), 3) self.assertEqual(len({call[1] for call in transport.submit_calls}), 1) self.assertEqual(sleeps, [0.1, 0.2]) self.assertEqual(runner.start(request()), record) self.assertEqual(len(transport.submit_calls), 3) def test_uncorrelated_409_is_safe_retryable_error(self): transport = FakeProcessingTransport( submit=[ ProcessingTransportError( "PROCESSING_ACTIVE_RUN_CONFLICT", retryable=True, active_run_conflict=True, ) ] ) runner = ProcessingRunner(transport, InMemoryProcessingLedger(), codec()) with self.assertRaises(ProcessingError) as raised: runner.start(request()) self.assertEqual(raised.exception.code, "PROCESSING_ACTIVE_RUN_CONFLICT") self.assertTrue(raised.exception.retryable) self.assertEqual(len(transport.submit_calls), 1) def test_rejected_start_is_recorded_failed_and_never_resubmitted(self): transport = FakeProcessingTransport( submit=[ ProcessingTransportError( "PROCESSING_REMOTE_REJECTED", retryable=False, ) ] ) ledger = InMemoryProcessingLedger() runner = ProcessingRunner(transport, ledger, codec()) with self.assertRaises(ProcessingError) as first: runner.start(request()) self.assertEqual(first.exception.code, "PROCESSING_REMOTE_REJECTED") record = ledger.get("job-processing-001", 1) self.assertIsNotNone(record) assert record is not None self.assertEqual(record.remote_status, "failed") self.assertEqual(record.failure_code, "PROCESSING_REMOTE_REJECTED") with self.assertRaises(ProcessingError) as replay: runner.start(request()) self.assertEqual(replay.exception.code, "PROCESSING_REMOTE_REJECTED") self.assertEqual(len(transport.submit_calls), 1) def test_exhausted_start_retries_are_recorded_failed(self): transport = FakeProcessingTransport( submit=[ ProcessingTransportError( "PROCESSING_REMOTE_UNAVAILABLE", retryable=True, ), ProcessingTransportError( "PROCESSING_REMOTE_UNAVAILABLE", retryable=True, ), ] ) ledger = InMemoryProcessingLedger() runner = ProcessingRunner( transport, ledger, codec(), max_transport_attempts=2, retry_delay_seconds=0, sleep=lambda _seconds: None, ) with self.assertRaises(ProcessingError) as raised: runner.start(request()) self.assertEqual(raised.exception.code, "PROCESSING_REMOTE_UNAVAILABLE") self.assertTrue(raised.exception.retryable) record = ledger.get("job-processing-001", 1) self.assertIsNotNone(record) assert record is not None self.assertEqual(record.remote_status, "failed") self.assertEqual(record.failure_code, "PROCESSING_REMOTE_UNAVAILABLE") def test_correlated_409_race_returns_the_persisted_run(self): ledger = InMemoryProcessingLedger() class RacingTransport: def submit(self, value, idempotency_key): ledger.bind_run(value.job_id, value.attempt_no, "run-raced-001", "running") raise ProcessingTransportError( "PROCESSING_ACTIVE_RUN_CONFLICT", retryable=True, active_run_conflict=True, ) def get(self, value, remote_run_id): raise AssertionError("not used") def cancel(self, value, remote_run_id): raise AssertionError("not used") runner = ProcessingRunner(RacingTransport(), ledger, codec()) record = runner.start(request()) self.assertEqual(record.remote_run_id, "run-raced-001") def test_poll_accepts_only_signed_correlated_result(self): processing_result = ProcessingResult.from_dict(result_payload()) signed = codec().sign(processing_result, issued_at=NOW, nonce="nonce-poll") transport = FakeProcessingTransport( submit=[RemoteRunSnapshot("run-processing-001", "running")], get=[ RemoteRunSnapshot("run-processing-001", "running"), RemoteRunSnapshot( "run-processing-001", "success", signed.decode("utf-8"), ), ], ) ledger = InMemoryProcessingLedger() sleeps: list[float] = [] runner = ProcessingRunner(transport, ledger, codec(), sleep=sleeps.append) runner.start(request()) completion = runner.poll( "job-processing-001", 1, poll_interval_seconds=0.5, timeout_seconds=5, ) self.assertEqual(completion.result, processing_result) self.assertFalse(completion.callback_replayed) self.assertEqual(sleeps, [0.5]) replay = runner.poll("job-processing-001", 1) self.assertTrue(replay.callback_replayed) def test_success_without_final_content_is_not_accepted(self): transport = FakeProcessingTransport( submit=[RemoteRunSnapshot("run-processing-001", "running")], get=[RemoteRunSnapshot("run-processing-001", "success", None)], ) runner = ProcessingRunner(transport, InMemoryProcessingLedger(), codec()) runner.start(request()) with self.assertRaises(ProcessingError) as raised: runner.poll("job-processing-001", 1) self.assertEqual(raised.exception.code, "PROCESSING_RESULT_MISSING") self.assertTrue(raised.exception.retryable) def test_unsigned_or_mismatched_final_content_is_rejected(self): unsigned = json.dumps(result_payload(), separators=(",", ":")) transport = FakeProcessingTransport( submit=[RemoteRunSnapshot("run-processing-001", "running")], get=[RemoteRunSnapshot("run-processing-001", "success", unsigned)], ) runner = ProcessingRunner(transport, InMemoryProcessingLedger(), codec()) runner.start(request()) with self.assertRaises(ProcessingError) as unsigned_error: runner.poll("job-processing-001", 1) self.assertEqual(unsigned_error.exception.code, "PROCESSING_SIGNATURE_INVALID") wrong_payload = result_payload(remote_run_id="run-other") wrong = codec().sign( ProcessingResult.from_dict(wrong_payload), issued_at=NOW, nonce="nonce-wrong", ) with self.assertRaises(ProcessingError) as mismatch: runner.accept_callback(wrong) self.assertEqual(mismatch.exception.code, "PROCESSING_RESULT_MISMATCH") def test_callback_replay_is_idempotent_but_changed_signature_conflicts(self): transport = FakeProcessingTransport( submit=[RemoteRunSnapshot("run-processing-001", "running")] ) ledger = InMemoryProcessingLedger() runner = ProcessingRunner(transport, ledger, codec()) runner.start(request()) result = ProcessingResult.from_dict(result_payload()) signed = codec().sign(result, issued_at=NOW, nonce="nonce-replay-1") first = runner.accept_callback(signed) second = runner.accept_callback(signed) self.assertFalse(first.callback_replayed) self.assertTrue(second.callback_replayed) resigned = codec().sign(result, issued_at=NOW, nonce="nonce-replay-2") with self.assertRaises(ProcessingError) as conflict: runner.accept_callback(resigned) self.assertEqual(conflict.exception.code, "PROCESSING_CALLBACK_CONFLICT") def test_poll_timeout_and_cancel_are_bounded(self): transport = FakeProcessingTransport( submit=[RemoteRunSnapshot("run-processing-001", "running")], get=[RemoteRunSnapshot("run-processing-001", "running")], cancel=[RemoteRunSnapshot("run-processing-001", "cancelling")], ) times = iter([0.0, 2.0]) runner = ProcessingRunner( transport, InMemoryProcessingLedger(), codec(), sleep=lambda _seconds: None, monotonic=lambda: next(times), ) runner.start(request()) with self.assertRaises(ProcessingError) as timeout: runner.poll( "job-processing-001", 1, poll_interval_seconds=0.1, timeout_seconds=1.0, ) self.assertEqual(timeout.exception.code, "PROCESSING_POLL_TIMEOUT") cancelled = runner.cancel("job-processing-001", 1) self.assertEqual(cancelled.remote_status, "cancelling") class FakeRemoteFiles: def __init__(self, objects: dict[str, bytes]) -> None: self.objects = objects self.calls: list[str] = [] def materialize(self, file_handle: str, destination: Path, max_bytes: int) -> None: self.calls.append(file_handle) if file_handle not in self.objects: raise ProcessingTransportError( "PROCESSING_REMOTE_FILE_MISSING", retryable=False ) value = self.objects[file_handle] if len(value) > max_bytes: raise ProcessingTransportError( "PROCESSING_REMOTE_FILE_TOO_LARGE", retryable=False ) destination.write_bytes(value) class ProcessingOutputRegistrarTests(unittest.TestCase): def _setup(self, root: Path): object_store = ManagedObjectStore( FilesystemObjectBackend(root / "objects", create=True) ) source_path = root / "upload.xml" source_path.write_bytes(b"") source = object_store.upload_committed( job_id="job-processing-001", attempt_no=1, role="source_xml", source=source_path, original_filename="source.xml", ).to_artifact_ref() files = FakeRemoteFiles( { "file_daily_report": b"daily", "file_result_json": b"result", "file_structured_result_json": b"structured", } ) return object_store, source, files def test_authenticated_handles_are_rechecked_committed_and_mapped_to_delivery(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) object_store, source, files = self._setup(root) registrar = ProcessingOutputRegistrar(files, object_store) registered = registrar.register( ProcessingResult.from_dict(result_payload()), source, ) self.assertEqual(registered.envelope.job_id, "job-processing-001") self.assertEqual(registered.envelope.status, "success") self.assertEqual( set(registered.output_objects), {"daily_report", "result_json", "structured_result_json"}, ) self.assertTrue( all(value.state == "committed" for value in registered.output_objects.values()) ) payload = json.loads(registered.raw_envelope) self.assertEqual(payload, registered.envelope.to_dict()) rendered = registered.raw_envelope.decode("utf-8") self.assertNotIn("file_daily_report", rendered) self.assertNotIn("local_path", rendered) def test_bad_remote_bytes_and_duplicate_handles_are_rejected(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) object_store, source, files = self._setup(root) files.objects["file_daily_report"] = b"wrong" registrar = ProcessingOutputRegistrar(files, object_store) with self.assertRaises(ProcessingError) as mismatch: registrar.register(ProcessingResult.from_dict(result_payload()), source) self.assertEqual( mismatch.exception.code, "PROCESSING_ARTIFACT_HASH_MISMATCH", ) duplicate_payload = result_payload() duplicate_payload["artifacts"]["result_json"]["file_handle"] = "file_daily_report" # type: ignore[index] files.objects["file_daily_report"] = b"daily" with self.assertRaises(ProcessingError) as duplicate: registrar.register(ProcessingResult.from_dict(duplicate_payload), source) self.assertEqual(duplicate.exception.code, "PROCESSING_ARTIFACT_INVALID") def test_remote_file_errors_are_safely_mapped(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) object_store, source, files = self._setup(root) del files.objects["file_result_json"] registrar = ProcessingOutputRegistrar(files, object_store) with self.assertRaises(ProcessingError) as missing: registrar.register(ProcessingResult.from_dict(result_payload()), source) self.assertEqual(missing.exception.code, "PROCESSING_REMOTE_FILE_MISSING") class FakeOpenAgentService: def __init__( self, response=None, error: Optional[BaseException] = None, trace_events: Optional[list[OpenAgentEvent]] = None, ) -> None: self.response = response or { "run_id": "run-processing-001", "status": "running", } self.error = error self.calls = [] self.trace_events = trace_events def send_message(self, *args, **kwargs): self.calls.append((args, kwargs)) if self.error is not None: raise self.error return self.response def stream_message(self, *args, **kwargs): self.calls.append((args, kwargs)) if self.error is not None: raise self.error yield from self.trace_events or [] def get_run(self, *_args, **_kwargs): if self.error is not None: raise self.error return self.response def cancel_run(self, *_args, **_kwargs): if self.error is not None: raise self.error return self.response class OpenAgentProcessingTransportTests(unittest.TestCase): def test_text_adapter_sends_only_canonical_request_and_safe_metadata(self): service = FakeOpenAgentService() transport = OpenAgentProcessingTransport(service) # type: ignore[arg-type] value = request() snapshot = transport.submit(value, "idem-001") self.assertEqual(snapshot.remote_run_id, "run-processing-001") args, kwargs = service.calls[0] self.assertNotIn(value.job_id, args[0]) self.assertEqual(json.loads(args[1]), value.to_dict()) self.assertEqual(kwargs["message_id"], "idem-001") self.assertEqual( kwargs["metadata"], { "contract_version": PROCESSING_REQUEST_VERSION, "processing_kind": "opera_daily", }, ) rendered = repr(service.calls[0]) self.assertNotIn("object_key", rendered) self.assertNotIn("local_path", rendered) def test_trace_adapter_returns_from_run_started_then_drains_safe_events(self): class MemoryTraceStore: def __init__(self) -> None: self.records = [] def append(self, job_id, record): self.records.append((job_id, dict(record))) service = FakeOpenAgentService( trace_events=[ OpenAgentEvent( event="trace", data={ "event": "run.started", "run_id": "run-trace-001", "ts": "2026-07-29T13:00:00Z", }, ), OpenAgentEvent( event="trace", data={ "event": "task.updated", "run_id": "run-trace-001", "ts": "2026-07-29T13:00:01Z", "data": {"summary": "开始处理任务"}, }, ), OpenAgentEvent( event="trace", data={ "event": "message.final", "run_id": "run-trace-001", "text": "submission_grant=must-not-escape", }, ), OpenAgentEvent( event="trace", data={ "event": "step.updated", "run_id": "run-trace-001", "ts": "2026-07-29T13:00:02Z", "data": {"summary": "工具完成 token=must-not-escape"}, }, ), OpenAgentEvent( event="trace", data={ "event": "run.completed", "run_id": "run-trace-001", "status": "success", "ts": "2026-07-29T13:00:03Z", }, ), OpenAgentEvent(event="end", data=None), ] ) trace_store = MemoryTraceStore() transport = OpenAgentProcessingTransport( service, # type: ignore[arg-type] trace_store=trace_store, ) value = request() snapshot = transport.submit(value, "idem-trace-001") transport.close(timeout_seconds=1.0) self.assertEqual(snapshot, RemoteRunSnapshot("run-trace-001", "running")) args, kwargs = service.calls[0] self.assertNotIn(value.job_id, args[0]) self.assertEqual(json.loads(args[1]), value.to_dict()) self.assertTrue(kwargs["include_trace"]) codes = [record[1]["code"] for record in trace_store.records] self.assertEqual( codes, [ "AGENT_RUN_STARTED", "AGENT_TASK_UPDATED", "AGENT_STEP_UPDATED", "AGENT_RUN_COMPLETED", "AGENT_TRACE_STREAM_ENDED", ], ) rendered = json.dumps(trace_store.records, ensure_ascii=False) self.assertNotIn("must-not-escape", rendered) def test_trace_submit_transport_failure_is_not_retried_without_server_deduplication(self): class MemoryTraceStore: def append(self, _job_id, _record): pass service = FakeOpenAgentService( error=OpenAgentTransportError("ambiguous after POST") ) transport = OpenAgentProcessingTransport( service, # type: ignore[arg-type] trace_store=MemoryTraceStore(), ) with self.assertRaises(ProcessingTransportError) as raised: transport.submit(request(), "idem-trace-ambiguous") self.assertEqual( raised.exception.code, "PROCESSING_REMOTE_SUBMISSION_AMBIGUOUS", ) self.assertFalse(raised.exception.retryable) def test_open_agent_409_429_and_timeout_are_normalized(self): errors = ( ( OpenAgentAPIError(409, "active", "raw"), "PROCESSING_ACTIVE_RUN_CONFLICT", True, ), ( OpenAgentAPIError(429, "limited", "raw"), "PROCESSING_REMOTE_RATE_LIMITED", False, ), ( OpenAgentTransportError("network timeout"), "PROCESSING_REMOTE_UNAVAILABLE", False, ), ) for error, expected_code, active_conflict in errors: with self.subTest(expected_code=expected_code): service = FakeOpenAgentService(error=error) transport = OpenAgentProcessingTransport(service) # type: ignore[arg-type] with self.assertRaises(ProcessingTransportError) as raised: transport.submit(request(), "idem-001") self.assertEqual(raised.exception.code, expected_code) self.assertTrue(raised.exception.retryable) self.assertEqual(raised.exception.active_run_conflict, active_conflict) if __name__ == "__main__": unittest.main(verbosity=2)