"""Unchanged 2.140 Store methods; diagnostic wrapper, not an old installation."""
from __future__ import annotations

import builtins
import json
from typing import Any

from muli.proactive import FrozenRouting
from muli.reply_origin import ReplyOrigin
from muli.store import Store


class PreviousInputConsumer(Store):
    def claim_inputs(
        self, session_id: str, run_id: str, *, limit: int
    ) -> builtins.list[dict[str, Any]]:
        """At a replay-safe boundary, atomically append inputs and advance delivery state."""
        result: builtins.list[dict[str, Any]] = []
        destination: tuple[str, ...] | None = None
        with self.db:
            rows = self.db.execute(
                "SELECT * FROM ingress WHERE session_id=? AND state='pending' ORDER BY seq LIMIT ?",
                (session_id, limit),
            ).fetchall()
            for row in rows:
                payload = json.loads(row["payload"])
                origin = ReplyOrigin.from_input(payload)
                routing = FrozenRouting.load(payload.get("attention_routing"))
                key = (
                    *origin.destination_key,
                    json.dumps(
                        [payload.get("attention_routing"), payload.get("reply_origin")],
                        sort_keys=True,
                        separators=(",", ":"),
                    )
                    if "attention_project" in payload
                    else routing.batch_key
                    if "attention_routing" in payload
                    else "",
                    json.dumps(
                        payload.get("channel_ownership"), sort_keys=True, separators=(",", ":")
                    ),
                    str(payload.get("channel_ownership_control", False)),
                    # A finite host proof freezes one independently revocable
                    # source. Equal routes must not merge different failures.
                    json.dumps(
                        payload["attention_memory_failure"],
                        sort_keys=True,
                        separators=(",", ":"),
                    )
                    if payload.get("principal") == "system:attention"
                    and "attention_memory_failure" in payload
                    else "",
                    json.dumps(
                        payload["attention_memory_watch"],
                        sort_keys=True,
                        separators=(",", ":"),
                    )
                    if payload.get("principal") == "system:attention"
                    and "attention_memory_watch" in payload
                    else "",
                    json.dumps(payload["attention_project"], sort_keys=True, separators=(",", ":"))
                    if "attention_project" in payload
                    else "",
                )
                if destination is not None and key != destination:
                    break  # Another recipient waits for the next replay-safe boundary.
                destination = key
                message = self._archive_input(
                    payload, received=row["received"], run_id=run_id, input_seq=row["seq"]
                )
                self.db.execute(
                    "UPDATE ingress SET state='delivered',delivered_seq=?,run_id=? WHERE seq=?",
                    (message["seq"], run_id, row["seq"]),
                )
                result.append(message)
        return result


    @staticmethod
    def input_archive(
        payload: dict[str, Any],
        *,
        received: float,
        run_id: str | None,
        input_seq: int | None = None,
    ) -> dict[str, Any]:
        """Project the original archive fields, shared by writes and evidence checks."""
        message = {
            "id": payload["id"],
            "session_id": payload["session_id"],
            "role": "user",
            "content": payload["content"],
            "created_at": received,
            "source": payload["source"],
            "principal": payload["principal"],
            "route": payload.get("route"),
            "attachment_ids": payload.get("attachment_ids", []),
            "delivery_status": "delivered",
            "run_id": run_id,
            "reply_origin": ReplyOrigin.from_input(payload).dump(),
        }
        if input_seq is not None:
            message["input_seq"] = input_seq
        if "task_control" in payload:
            message["task_control"] = payload["task_control"]
        if "attention_routing" in payload:
            message["attention_routing"] = FrozenRouting.load(payload["attention_routing"]).dump()
        if "attention_project" in payload:
            # Conversation already normalized trusted markers. Preserve even a
            # malformed stored marker and its original routing evidence so
            # Runtime rejects corruption rather than normalizing it to Web.
            message["attention_project"] = payload["attention_project"]
            message["reply_origin"] = payload.get("reply_origin")
            if "attention_routing" in payload:
                message["attention_routing"] = payload["attention_routing"]
        if "channel_ownership" in payload:
            message["channel_ownership"] = payload["channel_ownership"]
        if "channel_ownership_control" in payload:
            message["channel_ownership_control"] = payload["channel_ownership_control"]
        if "channel_reference" in payload:
            message["channel_reference"] = payload["channel_reference"]
        return message


    def _archive_input(
        self,
        payload: dict[str, Any],
        *,
        received: float,
        run_id: str | None,
        input_seq: int | None = None,
    ) -> dict[str, Any]:
        """Archive an input in the caller's transaction, preserving trusted provenance."""
        message = self.input_archive(payload, received=received, run_id=run_id, input_seq=input_seq)
        if self.get("message", payload["id"]) is not None:
            raise ValueError("输入消息 ID 已经存档，不能覆盖历史")
        self._put("message", message)
        return message


