새 답변과 매체 완료를 구분하는 합성 관찰기

"""Independent observer of declared synthetic snapshots. Never sends requests."""
import json
from copy import deepcopy


def fail(code):
    raise ValueError(code)


def label(value):
    if not isinstance(value, str) or not value or len(value)>100 or not all(c.isascii() and (c.isalnum() or c in "-_") for c in value):
        fail("LABEL")


def shape(value, keys):
    if not isinstance(value, dict) or set(value)!=set(keys):
        fail("SNAPSHOT_SHAPE")


def snapshot(value):
    shape(value,{"at","scope","busy","answers","media"})
    label(value["scope"])
    if type(value["at"]) is not int or value["at"]<0 or type(value["busy"]) is not bool:
        fail("SNAPSHOT_CLOCK_OR_BUSY")
    if not isinstance(value["answers"],list) or len(value["answers"])>100 or not isinstance(value["media"],list) or len(value["media"])>100:
        fail("SNAPSHOT_ROWS")
    ids=set()
    for a in value["answers"]:
        shape(a,{"id","text","waiting","failed","unavailable"}); label(a["id"])
        if a["id"]=="page" or a["id"] in ids:
            fail("ANSWER_ID")
        ids.add(a["id"])
        if not isinstance(a["text"],str) or len(a["text"])>20000 or any(type(a[k]) is not bool for k in ["waiting","failed","unavailable"]):
            fail("ANSWER_VALUE")
    ids=set()
    for m in value["media"]:
        shape(m,{"id","answer","kind","ordinal","ready","named"}); label(m["id"]); label(m["answer"])
        if m["id"] in ids:
            fail("MEDIA_ID")
        ids.add(m["id"])
        if m["kind"] not in ["image","video"] or type(m["ordinal"]) is not int or m["ordinal"]<0 or type(m["ready"]) is not bool or type(m["named"]) is not bool:
            fail("MEDIA_VALUE")
    return deepcopy(value)


class Observer:
    def __init__(self, baseline, *, target="text", timeout_ms=1000, allow_page_image=False):
        base=snapshot(baseline)
        if target not in ["text","image","video"] or type(timeout_ms) is not int or not 1<=timeout_ms<=600000 or type(allow_page_image) is not bool:
            fail("POLICY")
        if allow_page_image and target!="image":
            fail("PAGE_IMAGE_POLICY")
        self.scope=base["scope"]; self.target=target; self.allow_page_image=allow_page_image
        self.before={a["id"] for a in base["answers"]}
        self.before_image_count=sum(m["kind"]=="image" and m["ready"] for m in base["media"])
        self.before_image_ordinal=max([-1]+[m["ordinal"] for m in base["media"] if m["kind"]=="image" and m["ready"]])
        self.deadline=base["at"]+timeout_ms; self.last_at=base["at"]
        self.last_signature=None; self.repeats=0; self._terminal=None

    def _result(self, state, at, *, terminal=False, answer=None):
        value={"state":state,"at":at,"same_signature_repeats":self.repeats,
               "request_sent":False,"business_success_verified":False,
               "evidence":"caller-declared synthetic observations"}
        if answer is not None: value["answer"]=answer
        if terminal: self._terminal=deepcopy(value)
        return value

    def _reset(self):
        self.last_signature=None; self.repeats=0

    def _stable(self, value):
        sig=json.dumps(value,ensure_ascii=False,sort_keys=True,separators=(",",":"))
        if sig==self.last_signature: self.repeats+=1
        else: self.last_signature=sig; self.repeats=0
        return self.repeats>=2

    def result(self):
        return deepcopy(self._terminal)

    def observe(self, raw):
        if self._terminal is not None: fail("TERMINAL_OBSERVER")
        s=snapshot(raw)
        if s["scope"]!=self.scope: fail("SCOPE_CHANGED")
        if s["at"]<=self.last_at: fail("NON_INCREASING_CLOCK")
        self.last_at=s["at"]
        if s["at"]>=self.deadline:
            return self._result("timed_out",s["at"],terminal=True)
        if s["busy"]:
            self._reset(); return self._result("generating",s["at"])
        new=[a for a in s["answers"] if a["id"] not in self.before]
        answer=new[-1] if new else None
        if answer is None:
            if self.allow_page_image:
                images=[m for m in s["media"] if m["kind"]=="image" and m["ready"]]
                fresh=[m for m in images if m["answer"]=="page" and m["named"] and m["ordinal"]>self.before_image_ordinal]
                if len(images)>self.before_image_count and len(fresh)==1:
                    done=self._stable(["page-image",fresh[0]["id"],fresh[0]["ordinal"]])
                    return self._result("observed_complete" if done else "stabilizing",s["at"],terminal=done)
            self._reset(); return self._result("waiting_for_new_answer",s["at"])
        key=answer["id"]
        if answer["waiting"]:
            self._reset(); return self._result("waiting_for_user",s["at"],terminal=True,answer=key)
        if answer["failed"]:
            self._reset(); return self._result("response_failed",s["at"],terminal=True,answer=key)
        ready=[m for m in s["media"] if m["answer"]==key and m["kind"]==self.target and m["ready"]]
        if self.target!="text" and not ready:
            if answer["unavailable"]:
                done=self._stable(["unavailable",key,answer["text"].strip()])
                return self._result("media_unavailable" if done else "stabilizing_refusal",s["at"],terminal=done,answer=key)
            self._reset(); return self._result("waiting_for_media",s["at"],answer=key)
        if self.target=="text" and not answer["text"].strip():
            self._reset(); return self._result("waiting_for_text",s["at"],answer=key)
        # Include identity, unlike a text-only signature spanning two answer groups.
        done=self._stable([key,answer["text"].strip(),[(m["id"],m["kind"],m["ordinal"]) for m in ready]])
        return self._result("observed_complete" if done else "stabilizing",s["at"],terminal=done,answer=key)