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", "<RES_DETAIL"):
            self.assertNotIn(forbidden, rendered)

    def test_request_rejects_paths_urls_extra_fields_and_bad_attempt(self):
        baseline = request().to_dict()
        cases = []
        extra = dict(baseline)
        extra["object_key"] = "arr/jobs/private"
        cases.append(extra)
        path = dict(baseline)
        path["source_file_id"] = "/private/source.xml"
        cases.append(path)
        bad_attempt = dict(baseline)
        bad_attempt["attempt_no"] = 0
        cases.append(bad_attempt)
        for payload in cases:
            with self.subTest(payload=payload), self.assertRaises(ProcessingError):
                ProcessingRequest.from_dict(payload)

    def test_oss_program_message_matches_fetch_tool_attachment_contract(self):
        value = request(job_id="arrjob-source:message-001")
        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:
                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"<synthetic/>")
        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)
