A missing stop button should not by itself become proof that the requested AI result finished. After reading Northstar's waitAnswer functions and related tests, I built a small synthetic observer distinguishing a new answer from ready media. It sends no request and processes only the declared snapshots supplied to it.
Retain the answers already present at the start
The baseline keeps a scope, existing answer identities and ready-image count and ordinal. Later observations select the last answer absent from that initial identity set. A long, apparently finished earlier answer is not adopted as the new request's result. Scope is a synthetic organizing label rather than proof of login or permission.
Declared timestamps must strictly increase, preventing reuse of an identical snapshot to inflate stability. That is this example's condition, not a reproduction of every original observation system. Production integration needs a trusted connection between the request-start baseline and the current page. This module does not supply that binding or authenticate the caller's declarations.
Stability means an initial observation and two repeats
Busy state returns generating and clears the signature and repeat count. Missing new answers, required text or media return the appropriate waiting state. Once eligible, an initial signature is stored and must repeat twice more before observed_complete. This means three eligible observations, not completion after only two total reads.
The example includes answer identity in its signature, so identical text in another new answer cannot inherit stability. That is an independent design choice, not a claim that every original signature has the same fields. The code also performs no real polling interval or sleeping; its caller must supply observations. Stable declared data are evidence of the stated comparison, not permanent finality.
Image and video must be ready within the same answer
A text target needs a nonempty body. Image and video targets need ready media of the corresponding kind inside the same new answer identity; text may be empty. A video from another answer or a still-loading image does not qualify. An image cannot satisfy a video target simply because both are media.
An optional page-image exception handles a missing separate assistant heading. It must be explicitly enabled for an image target and requires no new answer, increased ready-image count, a newer ordinal and exactly one ready named page image. Ambiguous candidates are not resolved by choosing whichever one seems convenient. The policy makes the exceptional evidence inspectable.
Explore the implementation
Complete independent synthetic implementation
"""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)
Keep user waiting and refusal out of business-success claims
A new answer requesting user intervention ends as waiting_for_user; a failure declaration ends as response_failed. A stable unavailability declaration without required media becomes media_unavailable. The example does not classify natural language: it accepts synthetic boolean classifications that a real adapter would have to establish.
A refusal sentence can still be a completed text response, but that does not create the requested image. Every observed_complete therefore retains business_success_verified=False. Observed response completion and fulfillment of a business purpose remain different claims. A provider's inability response is not rewritten as an automation error or an invented generated artifact.
Thirty-two tests cover observations that must not complete
Tests cover old answers only, disappearance of stop without a new response, and stability reset after text or answer-identity changes. They distinguish ready same-answer media from other-answer, incomplete and wrong-kind media. Page-image name, count and ordinal requirements are checked separately. Invalid scope, clock and data shape are refused.
Thirty-two source tests run locally and the packaging procedure repeats them after ZIP extraction. Five records were generated through actual local function calls; the selector reads those results. No original browser or AI service was newly executed, and no existing product suite was rerun. Evidence records what declared observation was compared and by which code.
Timeout never sends another request
Reaching the declared deadline ends as timed_out. There is no real clock or background timer, so the observer does not wake itself when the caller stops feeding data. A terminal observer rejects further input and exposes detached result copies. It includes no durable journal, resume protocol or download capability, and request_sent is always False.
Code, tests, record generation and documentation permit modification and reuse. Operational use still needs a trusted site observer, request journal and actual artifact verification. I share this small state example so a delayed result can be understood through accumulated evidence before anyone treats resubmission as a solution.