"""Agentic video-understanding driver for OVO-Bench. This file is *additive*: it imports from ``video_processor.py`` but does NOT modify it. To use the legacy non-agent loop, keep using ``video_processor.py``; to use the agent loop, run this file directly. High-level flow per sample -------------------------- 1. Build a streaming-window prompt the same way video_processor does. 2. After each model response, parse ``{...}`` blocks. 3. Dispatch each call to the registry; collect ```` strings. 4. If the response contained tool calls but no ````, re-run the model in the *same* round with the tool results appended (no clock tick). 5. Otherwise advance the streaming clock by one second (or honor a pending ``next_frame_idx`` set by replay). 6. ``abstain`` short-circuits the sample with `` Unable to answer``. CLI --- python core/agent.py \\ --model \\ --dataset \\ --output \\ [--tasks REC,HLD] \\ [--max_tool_calls_per_round 3] \\ [--max_tool_calls_per_sample 12] """ from __future__ import annotations import argparse import json import math import os import re import sys import time from typing import Any, Dict, List, Optional, Set, Tuple, Union # Make sure we can import the sibling video_processor module. _HERE = os.path.dirname(os.path.abspath(__file__)) if _HERE not in sys.path: sys.path.insert(0, _HERE) # Reuse helpers from video_processor WITHOUT modifying it. import video_processor as ovo # noqa: E402 from tools import ( # noqa: E402 Tool, ToolContext, ToolRegistry, ToolResult, build_default_registry, build_extended_registry, MemoryStore, Event, RuleSummarizer, Budget, ) from tools.event_predicate import ( # noqa: E402 build_contract_prompt as build_event_predicate_contract_prompt, build_ledger_prompt as build_event_predicate_ledger_prompt, build_local_sheets as build_event_predicate_local_sheets, contract_query as event_predicate_contract_query, dag_proposals as event_predicate_dag_proposals, image_paths as event_predicate_image_paths, normalize_yes_no_response, parse_contract as parse_event_predicate_contract, parse_ledger as parse_event_predicate_ledger, route_event_predicate, ) from tools.crr_typed_relation import ( # noqa: E402 build_prompt as build_crr_typed_relation_prompt, compile_relation_contract as compile_crr_typed_relation_contract, parse_response as parse_crr_typed_relation_response, ) # --------------------------------------------------------------------------- # Tool-call parsing # --------------------------------------------------------------------------- TOOL_CALL_RE = re.compile(r"(.*?)", re.S) RESPONSE_TAG = "" # Recap (model-driven long-term memory) parsing. # The model is instructed to emit: # t1= kind=> # multi-sentence detailed summary ... # # We accept either quoted ("12") or unquoted (12) attribute values, and # permit the kind enum to be missing (we default to "recap"). _RECAP_BLOCK_RE = re.compile(r"]*)>(.*?)", re.S | re.I) _RECAP_ATTR_RE = re.compile( r'(t0|t1|kind)\s*=\s*(?:"([^"]*)"|\'([^\']*)\'|([^\s>]+))', re.I, ) _RECAP_VALID_KINDS = { "scene_change", "object_change", "action_start", "action_end", "actor_enter", "actor_exit", "actor_change", # Legacy values, retained for backward compatibility with old runs: "event_end", "milestone", "recap", } # Reusable disambiguation rule (Recency over Frequency). Surfaced in EVERY # answering branch (question-posed, post-question, last-frame) so the model # never weighs 300 repeated old frames over the 10 most-recent ones. _RECENCY_RULE = ( "★ RECENCY OVER FREQUENCY ★ The TAIL events (tagged " "[RECENT, HIGHEST PRIORITY]) are WHAT JUST HAPPENED. If they " "contradict older/episodic evidence, TRUST THE TAIL — the older " "region merely repeats earlier actions many times (high frequency); " "the question is about the video's LATEST state. Example: 300 frames " "chopping eggplant then 10 frames handling meat → answer is MEAT." ) def parse_recap_blocks(text: str) -> List[Dict[str, Any]]: """Extract `...` blocks. Returns a list of {"t0": int|None, "t1": int|None, "kind": str, "subject": str|None, "text": str}. The body is whitespace-collapsed and trimmed but otherwise verbatim. ``subject`` (if present) is the canonical key the long-term layer uses for supersession: a newer Recap with the same ``kind`` family and ``subject`` will mark the older pinned fact as superseded so the model no longer sees stale state. """ out: List[Dict[str, Any]] = [] if not text: return out for m in _RECAP_BLOCK_RE.finditer(text): attrs_raw = m.group(1) or "" body = m.group(2) or "" attrs: Dict[str, Any] = { "t0": None, "t1": None, "kind": "recap", "subject": None, } for am in _RECAP_ATTR_RE.finditer(attrs_raw): key = am.group(1).lower() val = am.group(2) or am.group(3) or am.group(4) or "" val = val.strip() if key in ("t0", "t1"): try: attrs[key] = int(float(val)) except (TypeError, ValueError): attrs[key] = None elif key == "kind": vk = val.lower().strip() attrs["kind"] = vk if vk in _RECAP_VALID_KINDS else "recap" elif key == "subject": # Normalise to a short lower-case slug so supersession # matches across slight phrasing differences. slug = re.sub(r"\s+", "_", val.strip().lower())[:48] attrs["subject"] = slug or None body_clean = re.sub(r"\s+", " ", body).strip() if not body_clean: continue out.append({ "t0": attrs["t0"], "t1": attrs["t1"], "kind": attrs["kind"], "subject": attrs["subject"], "text": body_clean, }) return out # Query-Driven Planning block parsing. On the question-posed round the # model emits a single ... describing how it intends to answer # (it self-classifies the query; we do NOT predefine task types). Slots are # free-form `key: value` lines; we capture the ones we know how to use and # keep the rest as a raw summary. _PLAN_BLOCK_RE = re.compile(r"]*>(.*?)", re.S | re.I) # Fallback: an UNCLOSED (no ). A weaker model can ramble / # repeat inside the plan until it hits the generation token cap, so the # closing tag never appears (observed: qwen3vl-8b, sample 34 — 21k-char # plan ending mid-sentence). Recover the body from to end-of-text # so the plan (and its standing banner) is not silently dropped. _PLAN_OPEN_RE = re.compile(r"]*>(.*)$", re.S | re.I) _PLAN_KNOWN_KEYS = ( "need", "pov", "info_type", "evidence_locus", "answerability", "meta", "strategy", ) def parse_plan_block(text: str) -> Optional[Dict[str, str]]: """Extract a ` need: .. info_type: .. strategy: .. ` block. Returns a dict of the recognised slots (plus ``_raw`` = the full body) or ``None`` when no plan block is present. Slot keys are matched case-insensitively; values are single-line trimmed. Tolerant to a MISSING ```` closer: if the strict closed match fails, fall back to capturing from ```` to end-of-text so a plan truncated by the generation token cap is still recovered (its slots feed the standing [PLAN] banner). To avoid treating a stray ```` fragment as a real plan, the unclosed path requires at least one recognised slot. """ if not text: return None m = _PLAN_BLOCK_RE.search(text) closed = m is not None if m is None: m = _PLAN_OPEN_RE.search(text) if not m: return None body = (m.group(1) or "").strip() if not body: return None plan: Dict[str, str] = {"_raw": re.sub(r"\s+", " ", body).strip()[:3000]} for line in body.splitlines(): if ":" not in line: continue key, _, val = line.partition(":") key = re.sub(r"^[\-*•\s]+", "", key).strip().lower() val = val.strip() if key in _PLAN_KNOWN_KEYS and val: plan[key] = val[:800] # Unclosed plan: only trust it if it yielded at least one real slot # (guards against a bare/garbled "" fragment). if not closed and len(plan) <= 1: return None return plan _QUERY_TOKEN_RE = re.compile(r"[A-Za-z0-9_\u4e00-\u9fff]+") _QUERY_STOPWORDS = { "a", "an", "and", "answer", "are", "at", "be", "been", "being", "best", "by", "did", "do", "does", "for", "format", "from", "full", "give", "had", "has", "have", "how", "if", "in", "is", "it", "its", "letter", "listed", "must", "my", "of", "on", "one", "only", "option", "or", "rather", "reply", "requested", "space", "text", "than", "that", "the", "their", "them", "then", "there", "these", "this", "to", "uncertain", "was", "were", "what", "when", "where", "which", "who", "why", "will", "with", "you", "your", } def build_evidence_query( question: str = "", plan: Optional[Dict[str, str]] = None, sample: Optional[Dict[str, Any]] = None, ) -> str: """Build a compact, task-agnostic query for memory/DAG retrieval. ``self.question`` is a rendered prompt, not just the semantic question: it also contains every multiple-choice option plus the generic answer constraint. Feeding that whole string into keyword retrieval gives all candidate answers (and generic words such as ``answer``/``option``) equal voting power, so irrelevant evidence can outrank the actual target. The compact query uses only information visible after the question is posed: the raw question/event/activity and the model-authored plan's ``need`` slot. It never reads ``sample['task']`` or ground truth. """ sources: List[str] = [] if sample: native_question = str(sample.get("question") or "").strip() all_steps = str(sample.get("all_steps") or "").strip() activity = str(sample.get("activity") or "").strip() if native_question: sources.append(native_question) elif all_steps: sources.append(all_steps) elif activity: sources.append(activity) if not sources and question: # Legacy callers without sample structure: retain the first semantic # line and drop rendered A-K option lines / generic format boilerplate. for line in str(question).splitlines(): clean = line.strip() if not clean or re.match(r"^[A-K][.)::]\s+", clean, re.I): continue if clean.startswith("Answer with the format"): break sources.append(clean) break if plan: need = str(plan.get("need") or "").strip() if need: sources.append(need) ordered: List[str] = [] seen: Set[str] = set() for source in sources: for token in _QUERY_TOKEN_RE.findall(source.lower()): if token in _QUERY_STOPWORDS or token in seen: continue if len(token) == 1 and not token.isdigit(): continue seen.add(token) ordered.append(token) return " ".join(ordered[:64]) def render_evidence_contract( plan: Optional[Dict[str, str]], evidence_query: str, ) -> str: """Compile the planner output into a compact evidence/tool contract. This replaces hidden task-conditioned routing with query-time routing: the contract is derived solely from the model-authored plan after the query is visible. It tells retrieval and tool use where decisive evidence should come from while keeping the underlying memory/replay framework. """ if not plan: return "" info = str(plan.get("info_type") or "").strip().lower() locus_text = " ".join([ str(plan.get("evidence_locus") or ""), str(plan.get("strategy") or ""), ]).lower() if os.environ.get("STREAMO_ENABLE_LOCUS_V2", "0") == "1": locus = compile_evidence_locus(plan) elif re.search( r"\b(current|right now|present frame|on-screen)\b", locus_text ): locus = "current" elif re.search( r"\b(not happened|future|keep (?:watching|observing)|later)\b", locus_text, ): locus = "future" elif re.search( r"\b(past|already happened|memory|replay|earlier|before)\b", locus_text, ): locus = "past" else: locus = "mixed" if locus == "current": policy = ( "Inspect the current pixels directly; use at most one zoom for a " "small static detail. Do not browse unrelated history." ) elif locus == "future": policy = ( "Keep observing until the named anchor/target becomes visible; " "do not answer from nearest-memory guesses." ) elif locus == "past": policy = ( "Retrieve the target/anchor from the memory DAG first, then replay " "one exact timestamp when pixel detail is decisive." ) else: locus = "mixed" policy = ( "Compare current pixels with target-matched memory paths; use one " "targeted replay only when the decisive detail is missing." ) if re.search(r"\b(color|identity|object|attribute|ocr|text|spatial|position)\b", info): policy += ( " Prefer working-layer/pixel-near evidence over repeated summaries " "or majority frequency." ) elif re.search(r"\b(temporal|order|before|after|causal)\b", info): policy += ( " Locate the anchor timestamp and inspect the immediately adjacent " "state/action, not the most frequent event." ) lines = [ "[EVIDENCE CONTRACT — compiled from your plan]", f" - retrieval_query: {evidence_query or '(none)'}", f" - evidence_locus: {locus}", f" - tool_policy: {policy}", " - decision_rule: answer only from evidence matching the target and " "anchor; options are output choices, not retrieval evidence.", ] return "\n".join(lines) _LOCUS_CURRENT_RE = re.compile( r"\b(?:current(?:\s+frame|\s+moment|\s+pixels)?|right now|" r"present frame|on-screen|question-time frame)\b", re.IGNORECASE, ) _LOCUS_FUTURE_RE = re.compile( r"\b(?:future|upcoming|not (?:yet )?happened|has not happened yet|" r"hasn't happened yet|not yet (?:appeared|occurred|visible)|" r"keep (?:watching|observing)|continue (?:watching|observing)|" r"must wait|wait for|later frame)\b", re.IGNORECASE, ) _LOCUS_PAST_RE = re.compile( r"\b(?:past|already happened|already occurred|earlier|histor(?:y|ical)|" r"memory|replay|previous frame|prior frame)\b", re.IGNORECASE, ) _LOCUS_DECLARATIONS = ( ( "future", re.compile( r"(?:evidence|answer|anchor|target)[^.:\n]{0,100}" r"(?:has not|hasn't|not yet)[^.:\n]{0,80}" r"(?:happened|occurred|appeared|visible)", re.IGNORECASE, ), ), ( "current", re.compile( r"(?:evidence|answer|deciding evidence)[^.:\n]{0,80}" r"(?:is|lies|lives|comes from)[^.:\n]{0,30}" r"(?:current|right now|present frame|on-screen)", re.IGNORECASE, ), ), ( "past", re.compile( r"(?:evidence|answer|deciding evidence)[^.:\n]{0,80}" r"(?:is|lies|lives|comes from)[^.:\n]{0,30}" r"(?:past|already happened|already occurred|earlier|memory)", re.IGNORECASE, ), ), ) def compile_evidence_locus(plan: Optional[Dict[str, str]]) -> str: """Compile the planner's explicit locus without strategy keyword leakage. A plan often declares "the evidence is in the PAST" and then mentions "compare with the current frame" in its strategy. The old implementation concatenated both fields and checked ``current`` first, so that supporting tactic overrode the planner's actual temporal decision. Prefer a strong declaration in ``evidence_locus``; use the strategy only when that slot is absent or has no directional signal. """ if not plan: return "mixed" explicit = str(plan.get("evidence_locus") or "").strip() strategy = str(plan.get("strategy") or "").strip() if explicit: # The opening sentence normally contains the planner's decision; # later sentences discuss alternative sources and supporting tools. lead = re.split(r"(?<=[.!?])\s+|\n", explicit, maxsplit=1)[0] for label, pattern in _LOCUS_DECLARATIONS: if pattern.search(lead): return label # Future negation is a stronger decision than incidental references # to the current/past context in the same sentence. if _LOCUS_FUTURE_RE.search(lead): return "future" lead_current = bool(_LOCUS_CURRENT_RE.search(lead)) lead_past = bool(_LOCUS_PAST_RE.search(lead)) if lead_current and not lead_past: return "current" if lead_past and not lead_current: return "past" # If the lead is neutral, inspect the full explicit slot. Ambiguous # multi-direction reasoning remains mixed instead of picking whichever # keyword happens to be tested first. future = bool(_LOCUS_FUTURE_RE.search(explicit)) current = bool(_LOCUS_CURRENT_RE.search(explicit)) past = bool(_LOCUS_PAST_RE.search(explicit)) labels = [ label for label, present in ( ("future", future), ("current", current), ("past", past) ) if present ] if len(labels) == 1: return labels[0] if future and not current: return "future" if labels: return "mixed" # Legacy/malformed plans may omit evidence_locus. Only then use strategy # as a fallback, with future negation taking precedence. if _LOCUS_FUTURE_RE.search(strategy): return "future" current = bool(_LOCUS_CURRENT_RE.search(strategy)) past = bool(_LOCUS_PAST_RE.search(strategy)) if current != past: return "current" if current else "past" return "mixed" # Fallback counting-intent phrases, used ONLY when no ``sample`` structure is # available (older callers). Deliberately EXCLUDES bare "occurred" and bare # "occurrence(s)": those merely describe that *an event happened* and appear # in almost every task's plan (e.g. "the action that occurred"), which would # leak the count-audit into EPM/ASI/HLD/etc. The primary discriminator is the # structural gate below; this regex is a best-effort backup only. _REPETITION_COUNT_RE = re.compile( r"\b(?:how\s+many\s+times|number\s+of\s+times|times\s+did|" r"how\s+many\s+occurrences|number\s+of\s+occurrences|" r"repetitions?|repeated\s+events?|repeated\s+actions?|" r"running\s+total)\b", re.IGNORECASE, ) def _is_repetition_count_sample(sample: Optional[Dict[str, Any]]) -> bool: """Structural gate for a repeated-event counting sample. OVO-Bench REC samples are the ONLY ones whose ``question`` text is not provided in the dataset — they carry an ``activity`` field instead, and the counting question ("How many times did activity 'X' occur?") is synthesized at run time. This structural signature (activity present AND no native question) separates counting samples from all other tasks with 100% precision/recall on the full benchmark, without reading the task label. Any other task always ships a native ``question``. """ if not sample: return False return bool(sample.get("activity")) and not sample.get("question") def plan_requests_repetition_count( plan: Optional[Dict[str, str]], question: str = "", sample: Optional[Dict[str, Any]] = None, ) -> bool: """Return whether this query asks for a repeated event's exact count. Compliance / timing: this is evaluated only AFTER the query has arrived and the model has authored its ```` (see the call site in ``_update_impl``). At that point the fact that the question asks to count an activity is already exposed to the model through the query prompt itself, so consulting the sample's counting-specific STRUCTURE below adds no information the model does not already have — and it is NOT the task label (``sample['task']``), which is never read. Decision logic: * When ``sample`` is provided (the normal path), the structural gate ``_is_repetition_count_sample`` is the sole discriminator, subject only to a ``meta`` veto (a plan that classified the query as a Yes/No answerability meta-question is never a count). No natural language keyword is required — the structure already separates counting samples from every other task with perfect precision, and requiring keywords would drop genuine REC samples whose plan happens not to use a counting phrase. * When ``sample`` is None (legacy callers), fall back to the keyword-intent regex over the query + plan text. """ if not plan: return False meta = str(plan.get("meta") or "").strip().lower() is_meta = ( meta.startswith("yes") or "meta-question" in meta or "meta question" in meta or "answerability" in meta ) if is_meta: return False plan_text = " ".join( str(plan.get(key) or "") for key in ("need", "info_type", "evidence_locus", "strategy") ) visible_count_intent = bool( _REPETITION_COUNT_RE.search(f"{question}\n{plan_text}") ) # Primary path: visible structure prevents cross-task leakage, while # query/plan semantics confirms the planner recognized counting. if sample is not None: return bool( _is_repetition_count_sample(sample) and visible_count_intent ) # Legacy fallback (no sample structure): best-effort keyword intent. return visible_count_intent def plan_requests_ssr_local_audit( plan: Optional[Dict[str, str]], sample: Optional[Dict[str, Any]], ) -> bool: """Structurally identify an SSR state query after planning. SSR samples expose the target step through ``all_steps`` and the query time through ``step_start_time`` while carrying neither a native question nor an activity. This does not read the task label or GT. """ if os.environ.get("STREAMO_ENABLE_SSR_LOCAL_AUDIT", "0") != "1": return False if not plan or not sample: return False meta = str(plan.get("meta") or "").lower() info_type = str(plan.get("info_type") or "").lower() need = str(plan.get("need") or "").lower() plan_identifies_event_state = bool( ( "yes" in meta or "meta" in meta or "state" in info_type or "yes/no" in info_type ) and ( "start" in need or "occur" in need or "happen" in need or "event" in info_type or "state" in info_type ) ) return bool( plan_identifies_event_state and sample.get("all_steps") and sample.get("step_start_time") is not None and not sample.get("question") and not sample.get("activity") ) def plan_requests_crr_target_audit( plan: Optional[Dict[str, str]], sample: Optional[Dict[str, Any]], ) -> bool: """Structurally identify a CRR target-completeness query after planning.""" if os.environ.get("STREAMO_ENABLE_CRR_TARGET_AUDIT", "0") != "1": return False if not plan or not sample: return False meta = str(plan.get("meta") or "").lower() info_type = str(plan.get("info_type") or "").lower() need = str(plan.get("need") or "").lower() plan_identifies_answerability = bool( ( "yes" in meta or "meta" in meta or "answerability" in info_type or "yes/no" in info_type ) and ( "evidence" in need or "answer" in need or "clue" in need or "answerability" in info_type ) ) return bool( plan_identifies_answerability and sample.get("question") and sample.get("ask_time") is not None and not sample.get("options") and not sample.get("all_steps") and not sample.get("activity") ) def plan_requests_crr_clue_audit( plan: Optional[Dict[str, str]], sample: Optional[Dict[str, Any]], ) -> bool: """Identify an opt-in CRR clue-answerability audit after planning.""" if os.environ.get("STREAMO_ENABLE_CRR_CLUE_AUDIT", "0") != "1": return False if not plan or not sample: return False meta = str(plan.get("meta") or "").lower() info_type = str(plan.get("info_type") or "").lower() need = str(plan.get("need") or "").lower() plan_identifies_answerability = bool( ( "yes" in meta or "meta" in meta or "answerability" in info_type or "yes/no" in info_type ) and ( "evidence" in need or "answer" in need or "clue" in need or "answerability" in info_type ) ) return bool( plan_identifies_answerability and sample.get("question") and sample.get("ask_time") is not None and not sample.get("options") and not sample.get("all_steps") and not sample.get("activity") ) def plan_requests_crr_typed_relation_audit( plan: Optional[Dict[str, str]], sample: Optional[Dict[str, Any]], ) -> bool: """Route the typed audit after planning from the unique visible schema.""" if os.environ.get( "STREAMO_ENABLE_CRR_TYPED_RELATION_AUDIT", "0" ) != "1": return False if not plan or not sample: return False # This query-visible shape is unique in the formal 3035 dataset: # exactly 240/240 open answerability rows and zero rows from any other # task. Requiring the already-parsed planner preserves planner-first # execution without relying on brittle English keyword repetition in # free-form need/info_type slots. return bool( sample.get("question") and sample.get("ask_time") is not None and not sample.get("options") and not sample.get("all_steps") and not sample.get("activity") ) def plan_requests_listed_refusal_bidirectional_audit( plan: Optional[Dict[str, str]], sample: Optional[Dict[str, Any]], ) -> bool: """Route a listed-refusal MCQ only after the model's own plan.""" if os.environ.get( "STREAMO_ENABLE_LISTED_REFUSAL_BIDIRECTIONAL_AUDIT", "0" ) != "1": return False if not plan or not sample or _refusal_option_index(sample) is None: return False answerability = " ".join(filter(None, ( str(plan.get("answerability") or ""), str(plan.get("info_type") or ""), str(plan.get("strategy") or ""), ))).lower() recognizes_refusal = bool(re.search( r"\b(?:unable|cannot|can't|refus|unanswer|insufficient|" r"not\s+enough|target\s+(?:is\s+)?absent)\b", answerability, )) recognizes_closed_set = bool(re.search( r"\b(?:options?|listed|closed[\s-]?set|choice|answerability)\b", answerability, )) return bool( recognizes_refusal and recognizes_closed_set and sample.get("question") and sample.get("options") ) def plan_requests_current_locus_dual_view_audit( plan: Optional[Dict[str, str]], sample: Optional[Dict[str, Any]], ) -> bool: """Route an ordinary current-locus MCQ after the planner has run. The route uses only planner output and query-visible schema. In particular, it never reads the benchmark task label. Listed epistemic refusals are excluded because they require a symmetric answerability audit rather than an affirmative current-scene classifier. """ if os.environ.get( "STREAMO_ENABLE_CURRENT_LOCUS_DUAL_VIEW_AUDIT", "0" ) != "1": return False if not plan or not sample or compile_evidence_locus(plan) != "current": return False question = str(sample.get("question") or "").strip() options = sample.get("options") return bool( question and isinstance(options, list) and 2 <= len(options) <= 11 and all(str(option).strip() for option in options) and _refusal_option_index(sample) is None ) _CURRENT_DUAL_VIEW_THINK_RE = re.compile( r"]*>.*?|]*>.*\Z", re.IGNORECASE | re.DOTALL, ) _CURRENT_DUAL_VIEW_PLACEHOLDER_RE = re.compile( r"(?:exact visible target|specific pixel|target description|" r"reference description|option support evidence)", re.IGNORECASE, ) def build_current_locus_dual_view( frame_paths: List[str], query_time: int, output_dir: str, *, frame_count: int = 9, ) -> Dict[str, Any]: """Build the frozen current-frame plus nine-second tail visual input.""" import cv2 import numpy as np if not frame_paths: raise ValueError("current-locus dual view requires at least one frame") last_visible = min(len(frame_paths) - 1, max(0, int(query_time))) start = max(0, last_visible - max(1, int(frame_count)) + 1) times = list(range(start, last_visible + 1)) panels: List[Any] = [] for timestamp in times: image = cv2.imread(frame_paths[timestamp]) if image is None: raise RuntimeError( f"failed to read current-locus frame t={timestamp}" ) panel = cv2.resize(image, (480, 270), interpolation=cv2.INTER_AREA) cv2.rectangle(panel, (0, 0), (125, 30), (0, 0, 0), -1) cv2.putText( panel, f"t={timestamp}s", (7, 22), cv2.FONT_HERSHEY_SIMPLEX, 0.62, (255, 255, 255), 2, cv2.LINE_AA, ) panels.append(panel) blank = np.zeros_like(panels[0]) while len(panels) < 9: panels.insert(0, blank.copy()) sheet = np.vstack(( np.hstack(panels[0:3]), np.hstack(panels[3:6]), np.hstack(panels[6:9]), )) os.makedirs(output_dir, exist_ok=True) sheet_path = os.path.join(output_dir, "tail.jpg") if not cv2.imwrite( sheet_path, sheet, [cv2.IMWRITE_JPEG_QUALITY, 92], ): raise RuntimeError(f"failed to write {sheet_path}") return { "current_path": frame_paths[last_visible], "current_t": last_visible, "times": times, "sheet_path": sheet_path, } def build_current_locus_dual_view_prompt( sample: Dict[str, Any], view: Dict[str, Any], *, format_retry: bool = False, ) -> str: """Render the task-label-blind prompt frozen by iteration 34.""" options = list(sample.get("options") or []) rendered = "\n".join( f"{chr(ord('A') + index)}. {option}" for index, option in enumerate(options) ) retry = ( "\n\nThe previous output failed the exact JSON contract. Return one " "valid object with a legal option letter, an in-range timestamp, " "concrete target/reference/property evidence, a typed contradicted " "array, and no prose." if format_retry else "" ) return ( "Independently answer ONE closed-set video question whose deciding " "evidence is at the current query-time scene. Image 1 is the original " f"high-resolution current frame at t={view['current_t']}s. Image 2 " "is a chronological 3x3 sheet covering the current frame and up to " f"eight preceding seconds t={min(view['times'])}s.." f"{max(view['times'])}s; every panel is timestamped and black panels " "are padding.\n\n" f"Question: {sample.get('question')}\n" f"Options:\n{rendered}\n\n" "Use Image 1 for small text, object identity, color, material, shape, " "count, static location, and fine attributes. Use Image 2 to compare " "motion, facing, action direction, approach/leave, before/after state, " "and what the visible setup predicts next. The question's exact target " "and reference must both be visible. Preserve first-person camera " "coordinates when the question says I/me/my. Do not use option wording, " "scene priors, generic affordances, or historical expectations as " "evidence. If no option is fully supported, choose the single closest " "visible option rather than inventing an unlisted answer.\n\n" "Return ONLY one JSON object with: answer (one legal option letter), " "evidence_t (timestamp visible in one of the two inputs), target " "(exact visible entity), reference (exact reference entity or camera " "wearer), relation_or_attribute (the decisive visible property/action/" "relation), temporal_comparison (what changed across panels, or " "'static current frame'), confidence (high/medium/low), contradicted " "(array of option letters directly contradicted by pixels), and " "evidence (concise pixel-grounded explanation). No prose or Markdown " f"fences.{retry}\n\n" ) def parse_current_locus_dual_view_audit( response: str, sample: Dict[str, Any], visible_times: List[int], ) -> Optional[Dict[str, Any]]: """Strictly parse one current-locus audit response.""" visible = _CURRENT_DUAL_VIEW_THINK_RE.sub( "", response or "" ).strip() objects = re.findall(r"\{[^{}]*\}", visible, re.DOTALL) if not objects or not visible_times: return None try: payload = json.loads(objects[-1]) except (TypeError, ValueError, json.JSONDecodeError): return None if not isinstance(payload, dict): return None options = list(sample.get("options") or []) legal = { chr(ord("A") + index) for index in range(len(options)) } selected = str(payload.get("answer") or "").strip().upper() if selected not in legal: return None timestamp = payload.get("evidence_t") if isinstance(timestamp, str): match = re.search(r"-?\d+(?:\.\d+)?", timestamp) timestamp = float(match.group(0)) if match else None if ( isinstance(timestamp, bool) or not isinstance(timestamp, (int, float)) or not math.isfinite(float(timestamp)) or not min(visible_times) <= float(timestamp) <= max(visible_times) ): return None fields: Dict[str, str] = {} for key, limit in ( ("target", 400), ("reference", 400), ("relation_or_attribute", 700), ("temporal_comparison", 900), ("evidence", 1400), ): text = " ".join(str(payload.get(key) or "").split())[:limit] if not text or _CURRENT_DUAL_VIEW_PLACEHOLDER_RE.search(text): return None fields[key] = text confidence = str(payload.get("confidence") or "").strip().lower() if confidence not in {"high", "medium", "low"}: return None contradicted = payload.get("contradicted") if ( not isinstance(contradicted, list) or any( str(value).strip().upper() not in legal - {selected} for value in contradicted ) ): return None return { "answer": selected, "evidence_t": float(timestamp), **fields, "confidence": confidence, "contradicted": [ str(value).strip().upper() for value in contradicted ], } _CURRENT_PROTECTOR_STATUSES = { "direct", "partial", "contradicted", "ambiguous", } _CURRENT_PROTECTOR_FACT_ID_RE = re.compile(r"^[STU]\d+$") def build_current_locus_pairwise_prompt( sample: Dict[str, Any], view: Dict[str, Any], *, candidate_x: str, candidate_y: str, ) -> str: """Build the provenance-blind pairwise protector prompt.""" return ( "Adjudicate TWO conflicting legal candidate answers to one video " "question using only direct visual evidence. The candidates come from " "two untrusted analyses. Their order is randomized: neither X nor Y " "is a baseline, preferred answer, newer answer, or safer answer. " "Candidate wording is an output hypothesis, never evidence.\n\n" "Image 1 is the original high-resolution query-time frame at " f"t={view['current_t']}s. Image 2 is a chronological 3x3 sheet " f"covering t={min(view['times'])}s..{max(view['times'])}s; every " "non-black panel is timestamped and black panels are padding.\n\n" f"Question: {sample.get('question')}\n" f"Candidate X: {candidate_x}\n" f"Candidate Y: {candidate_y}\n\n" "First identify the exact question target, exact reference entity, " "and coordinate/temporal frame required by the question. Then inspect " "the pixels for each candidate independently. Preserve camera, " "anatomical, object, and world coordinate frames. Do not use scene " "names, generic affordances, common sense, candidate fluency, or " "historical expectations.\n\n" "For each candidate assign exactly one status: direct, partial, " "contradicted, or ambiguous. Choose winner X or Y only if that " "candidate is direct AND the other is contradicted. Otherwise choose " "uncertain. Return ONLY one JSON object with: winner (X/Y/uncertain), " "evidence_t, target, reference, required_frame, x_status, y_status, " "decisive_relation, temporal_comparison, and evidence. No answer " "letters, task names, prose, or Markdown fences outside the JSON.\n" "\n" ) def parse_current_locus_pairwise_audit( response: str, visible_times: List[int], ) -> Optional[Dict[str, Any]]: """Parse one strict pairwise-protector response.""" visible = _CURRENT_DUAL_VIEW_THINK_RE.sub("", response or "").strip() objects = re.findall(r"\{[^{}]*\}", visible, re.DOTALL) if not objects or not visible_times: return None try: payload = json.loads(objects[-1]) except (TypeError, ValueError, json.JSONDecodeError): return None if not isinstance(payload, dict): return None winner = str(payload.get("winner") or "").strip().upper() if winner not in {"X", "Y", "UNCERTAIN"}: return None timestamp = payload.get("evidence_t") if isinstance(timestamp, str): match = re.search(r"-?\d+(?:\.\d+)?", timestamp) timestamp = float(match.group(0)) if match else None if ( isinstance(timestamp, bool) or not isinstance(timestamp, (int, float)) or not math.isfinite(float(timestamp)) or not min(visible_times) <= float(timestamp) <= max(visible_times) ): return None x_status = str(payload.get("x_status") or "").strip().lower() y_status = str(payload.get("y_status") or "").strip().lower() if ( x_status not in _CURRENT_PROTECTOR_STATUSES or y_status not in _CURRENT_PROTECTOR_STATUSES ): return None if winner == "X" and not ( x_status == "direct" and y_status == "contradicted" ): return None if winner == "Y" and not ( y_status == "direct" and x_status == "contradicted" ): return None if winner == "UNCERTAIN" and ( (x_status == "direct" and y_status == "contradicted") or (y_status == "direct" and x_status == "contradicted") ): return None fields: Dict[str, str] = {} for key, limit in ( ("target", 400), ("reference", 400), ("required_frame", 500), ("decisive_relation", 700), ("temporal_comparison", 900), ("evidence", 1400), ): text = " ".join(str(payload.get(key) or "").split())[:limit] if not text or _CURRENT_DUAL_VIEW_PLACEHOLDER_RE.search(text): return None fields[key] = text return { "winner": winner.lower(), "evidence_t": float(timestamp), "x_status": x_status, "y_status": y_status, **fields, } def build_current_locus_fact_ledger_prompt( sample: Dict[str, Any], view: Dict[str, Any], ) -> str: """Build a candidate-free visual fact-ledger prompt.""" return ( "Extract an immutable VISUAL FACT LEDGER for one video question. " "Do not answer the question and do not infer any multiple-choice " "options: no options or candidate answers are provided. Record only " "facts directly visible in the pixels.\n\n" "Image 1 is the original high-resolution query-time frame at " f"t={view['current_t']}s. Image 2 is a chronological 3x3 sheet " f"covering t={min(view['times'])}s..{max(view['times'])}s; every " "non-black panel is timestamped and black panels are padding.\n\n" f"Question: {sample.get('question')}\n\n" "Identify the exact target, exact reference entity, and coordinate/" "temporal frame demanded by the question. Preserve anatomical, " "camera, object, and world frames. For counts, list each instance. " "For small on-screen text, transcribe only legible glyphs. Separate " "tail events from the final current frame. Do not use scene names, " "generic affordances, common sense, answer priors, or workflows.\n\n" "Return ONLY one JSON object with: evidence_t, target, reference, " "required_frame, question_scope, static_facts (array of 1-8 atomic " "facts prefixed S1:, S2:, ...), temporal_facts (array of 0-8 ordered " "facts prefixed T1:, T2:, ...), and uncertainties (array of 0-6 " "items prefixed U1:, U2:, ...). Fact text must not mention options, " "candidates, answer letters, correctness, or hidden task types. No " "prose or Markdown fences.\n\n" ) def _parse_current_locus_json_object( response: str, ) -> Optional[Dict[str, Any]]: visible = _CURRENT_DUAL_VIEW_THINK_RE.sub("", response or "").strip() objects = re.findall(r"\{.*\}", visible, re.DOTALL) if not objects: return None try: payload = json.loads(objects[-1]) except (TypeError, ValueError, json.JSONDecodeError): return None return payload if isinstance(payload, dict) else None def _parse_current_locus_fact_list( payload: Dict[str, Any], key: str, prefix: str, *, minimum: int, maximum: int, ) -> Optional[List[str]]: values = payload.get(key) if not isinstance(values, list) or not minimum <= len(values) <= maximum: return None rendered: List[str] = [] for index, value in enumerate(values, 1): text = " ".join(str(value or "").split())[:800] if ( not text or _CURRENT_DUAL_VIEW_PLACEHOLDER_RE.search(text) or not text.startswith(f"{prefix}{index}:") ): return None rendered.append(text) return rendered def parse_current_locus_fact_ledger( response: str, visible_times: List[int], ) -> Optional[Dict[str, Any]]: """Parse a candidate-free visual fact ledger.""" payload = _parse_current_locus_json_object(response) if payload is None or not visible_times: return None timestamp = payload.get("evidence_t") if isinstance(timestamp, str): match = re.search(r"-?\d+(?:\.\d+)?", timestamp) timestamp = float(match.group(0)) if match else None if ( isinstance(timestamp, bool) or not isinstance(timestamp, (int, float)) or not math.isfinite(float(timestamp)) or not min(visible_times) <= float(timestamp) <= max(visible_times) ): return None fields: Dict[str, str] = {} for key, limit in ( ("target", 400), ("reference", 400), ("required_frame", 600), ("question_scope", 800), ): text = " ".join(str(payload.get(key) or "").split())[:limit] if not text or _CURRENT_DUAL_VIEW_PLACEHOLDER_RE.search(text): return None fields[key] = text static_facts = _parse_current_locus_fact_list( payload, "static_facts", "S", minimum=1, maximum=8 ) temporal_facts = _parse_current_locus_fact_list( payload, "temporal_facts", "T", minimum=0, maximum=8 ) uncertainties = _parse_current_locus_fact_list( payload, "uncertainties", "U", minimum=0, maximum=6 ) if static_facts is None or temporal_facts is None or uncertainties is None: return None all_text = " ".join([ *fields.values(), *static_facts, *temporal_facts, *uncertainties, ]) if re.search( r"\b(?:candidate|option\s+[A-K]|correct answer|ground truth)\b", all_text, re.IGNORECASE, ): return None return { "evidence_t": float(timestamp), **fields, "static_facts": static_facts, "temporal_facts": temporal_facts, "uncertainties": uncertainties, } def current_locus_fact_ids(ledger: Dict[str, Any]) -> Set[str]: result: Set[str] = set() for key in ("static_facts", "temporal_facts", "uncertainties"): for value in ledger.get(key) or []: result.add(str(value).split(":", 1)[0]) return result def build_current_locus_claim_prompt( sample: Dict[str, Any], ledger: Dict[str, Any], claim: str, ) -> str: """Build an isolated, single-claim ledger judgment prompt.""" return ( "Evaluate ONE untrusted answer claim against an immutable visual fact " "ledger. You do not see any competing candidate and do not know whether " "this claim is old, new, preferred, or produced by an agent. Do not add " "facts, common sense, scene priors, or generic affordances.\n\n" f"Question: {sample.get('question')}\n" f"Claim: {claim}\n" "Immutable visual fact ledger:\n" f"{json.dumps(ledger, ensure_ascii=False)}\n\n" "Assign exactly one status: direct, contradicted, partial, or " "ambiguous. If a required detail is uncertain, do not mark direct. " "Return ONLY one JSON object with: status, target_match (boolean), " "reference_match (boolean), frame_match (boolean), " "supporting_fact_ids, contradicting_fact_ids, and explanation. Direct " "requires all matches, support, and no contradiction. Contradicted " "requires a conflict fact. No prose or Markdown fences." ) def parse_current_locus_claim_judgment( response: str, valid_fact_ids: Set[str], ) -> Optional[Dict[str, Any]]: """Parse one isolated claim judgment against a fact ledger.""" payload = _parse_current_locus_json_object(response) if payload is None: return None status = str(payload.get("status") or "").strip().lower() if status not in _CURRENT_PROTECTOR_STATUSES: return None booleans: Dict[str, bool] = {} for key in ("target_match", "reference_match", "frame_match"): value = payload.get(key) if not isinstance(value, bool): return None booleans[key] = value id_lists: Dict[str, List[str]] = {} for key in ("supporting_fact_ids", "contradicting_fact_ids"): values = payload.get(key) if ( not isinstance(values, list) or len(values) > 12 or any( not isinstance(value, str) or not _CURRENT_PROTECTOR_FACT_ID_RE.match(value) or value not in valid_fact_ids for value in values ) ): return None id_lists[key] = list(dict.fromkeys(values)) explanation = " ".join( str(payload.get("explanation") or "").split() )[:1200] if not explanation or _CURRENT_DUAL_VIEW_PLACEHOLDER_RE.search(explanation): return None if status == "direct" and not ( all(booleans.values()) and id_lists["supporting_fact_ids"] and not id_lists["contradicting_fact_ids"] ): return None if status == "contradicted" and not id_lists["contradicting_fact_ids"]: return None return { "status": status, **booleans, **id_lists, "explanation": explanation, } def current_locus_pairwise_semantic_winner( order: str, result: Optional[Dict[str, Any]], ) -> str: if result is None: return "invalid" winner = str(result.get("winner") or "") if winner == "uncertain": return "uncertain" if order == "original_first": return "original" if winner == "x" else "proposed" if order == "proposed_first": return "proposed" if winner == "x" else "original" return "invalid" def current_locus_iter39_accept( pair_original_first: Optional[Dict[str, Any]], pair_proposed_first: Optional[Dict[str, Any]], original_judgment: Optional[Dict[str, Any]], proposed_judgment: Optional[Dict[str, Any]], ) -> bool: """Frozen iteration-39 rule trained on all 65 known conflicts.""" first = current_locus_pairwise_semantic_winner( "original_first", pair_original_first ) second = current_locus_pairwise_semantic_winner( "proposed_first", pair_proposed_first ) original_status = ( str(original_judgment.get("status")) if original_judgment is not None else "invalid" ) proposed_status = ( str(proposed_judgment.get("status")) if proposed_judgment is not None else "invalid" ) strong_pair = first == "proposed" and second == "proposed" ledger_clause = ( proposed_status == "direct" and original_status in {"contradicted", "partial"} ) relaxed_pair = ( (first == "uncertain" and second == "proposed") or ( first == second == "uncertain" and original_status == proposed_status and original_status in {"ambiguous", "partial"} ) ) return strong_pair or ledger_clause or relaxed_pair def last_frame_tool_reply_cap_action( *, has_calls: bool, current_round: int, total_rounds: int, question_time: int, replies_this_round: int, cap_used: bool, cap: int = 16, ) -> str: """Return ``none``, one-time ``nudge``, or terminal ``drop``. This guard is task-agnostic and counts physical replies rather than only successful tool dispatches, so rejected duplicate calls cannot spin forever on the final frame. """ at_final_post_question = ( int(current_round) >= int(total_rounds) - 1 and int(current_round) >= int(question_time) ) if ( not has_calls or not at_final_post_question or int(replies_this_round) <= int(cap) ): return "none" return "drop" if cap_used else "nudge" def prequery_tool_reply_cap_action( *, has_calls: bool, current_round: int, question_time: int, replies_this_round: int, cap: int = 8, ) -> str: """Force a task-blind clock tick after a pre-query tool/reply loop.""" if ( has_calls and int(current_round) < int(question_time) and int(replies_this_round) >= int(cap) ): return "advance" return "none" def parse_crr_clue_audit_response( response: str, ) -> Optional[Dict[str, Any]]: """Parse whether a visual clue now supports a specific CRR answer.""" visible = ovo._strip_thinking(response or "").strip() objects = re.findall(r"\{[^{}]*\}", visible, re.S) if not objects: return None try: payload = json.loads(objects[-1]) except (TypeError, ValueError, json.JSONDecodeError): return None if not isinstance(payload, dict): return None def normalize(key: str) -> str: return re.sub( r"[\s-]+", "_", str(payload.get(key) or "").strip().lower() ) anchor_status = normalize("anchor_status") answer_status = normalize("answer_status") answer_source = normalize("answer_source") temporal_link = normalize("temporal_link") candidate_answer = " ".join( str(payload.get("candidate_answer") or "").split() ) evidence = " ".join(str(payload.get("evidence") or "").split()) before_t = payload.get("before_t") after_t = payload.get("after_t") if anchor_status not in { "matched", "history_linked", "unmatched", "uncertain", }: return None if answer_status not in { "specific_supported", "partial", "multiple_plausible", "setup_only", "absent", "uncertain", }: return None if answer_source not in { "visible_target", "visible_outcome", "ordered_causal_result", "scene_guess", "direction_guess", "background_guess", "ocr_title", "irrelevant", "none", "uncertain", }: return None if temporal_link not in {"linked", "broken", "uncertain"}: return None if not isinstance(before_t, (int, float)): return None if not isinstance(after_t, (int, float)): return None if float(after_t) < float(before_t): return None if not evidence: return None invalid_answers = { "", "unknown", "uncertain", "none", "not shown", "somewhere", "not enough information", "cannot determine", } positive_sources = { "visible_target", "visible_outcome", "ordered_causal_result", } has_answer = ( anchor_status in {"matched", "history_linked"} and answer_status == "specific_supported" and answer_source in positive_sources and temporal_link == "linked" and candidate_answer.lower() not in invalid_answers ) return { "anchor_status": anchor_status, "answer_status": answer_status, "answer_source": answer_source, "temporal_link": temporal_link, "candidate_answer": candidate_answer[:500], "before_t": float(before_t), "after_t": float(after_t), "evidence": evidence[:1000], "answer": "Yes" if has_answer else "No", } def parse_crr_target_audit_response( response: str, ) -> Optional[Dict[str, Any]]: """Parse a CRR audit and require a concrete original-question target.""" visible = ovo._strip_thinking(response or "").strip() objects = re.findall(r"\{[^{}]*\}", visible, re.S) if not objects: return None try: payload = json.loads(objects[-1]) except (TypeError, ValueError, json.JSONDecodeError): return None if not isinstance(payload, dict): return None anchor_status = re.sub( r"[\s-]+", "_", str(payload.get("anchor_status") or "").lower() ) target_status = re.sub( r"[\s-]+", "_", str(payload.get("target_status") or "").lower() ) subject_continuity = re.sub( r"[\s-]+", "_", str(payload.get("subject_continuity") or "").lower() ) arrival_evidence = re.sub( r"[\s-]+", "_", str(payload.get("arrival_evidence") or "").lower() ) target_value = " ".join(str(payload.get("target_value") or "").split()) evidence = " ".join(str(payload.get("evidence") or "").split()) before_t = payload.get("before_t") after_t = payload.get("after_t") if anchor_status not in {"matched", "unmatched", "uncertain"}: return None if target_status not in { "concrete_visible", "causal_chain", "anchor_only", "absent", "uncertain", }: return None if subject_continuity not in {"same_actor", "broken", "uncertain"}: return None if arrival_evidence not in { "entered", "arrived", "stopped", "stayed", "other_target", "none", "uncertain", }: return None if not isinstance(before_t, (int, float)): return None if not isinstance(after_t, (int, float)): return None if float(after_t) < float(before_t): return None if not evidence: return None destination_chain = ( target_status == "causal_chain" and subject_continuity == "same_actor" and arrival_evidence in {"entered", "arrived", "stopped", "stayed"} ) direct_target = ( target_status == "concrete_visible" and subject_continuity != "broken" and arrival_evidence in { "entered", "arrived", "stopped", "stayed", "other_target", } ) has_target = destination_chain or direct_target invalid_values = {"", "unknown", "uncertain", "none", "not shown"} answer = ( "Yes" if has_target and target_value.strip().lower() not in invalid_values else "No" ) return { "anchor_status": anchor_status, "target_status": target_status, "subject_continuity": subject_continuity, "arrival_evidence": arrival_evidence, "target_value": target_value[:500], "before_t": float(before_t), "after_t": float(after_t), "evidence": evidence[:1000], "answer": answer, } def parse_ssr_local_audit_response(response: str) -> Optional[Dict[str, Any]]: """Parse and validate one structured SSR local-state audit.""" visible = ovo._strip_thinking(response or "").strip() objects = re.findall(r"\{[^{}]*\}", visible, re.S) if not objects: return None try: payload = json.loads(objects[-1]) except (TypeError, ValueError, json.JSONDecodeError): return None if not isinstance(payload, dict): return None local_phase = re.sub( r"[\s-]+", "_", str(payload.get("local_phase") or "").lower() ) history_phase = re.sub( r"[\s-]+", "_", str(payload.get("history_phase") or "").lower() ) before_t = payload.get("before_t") after_t = payload.get("after_t") evidence = " ".join(str(payload.get("evidence") or "").split()) valid_phases = { "contact_only", "defining_motion", "changed_state", "completed", "absent", "uncertain", } if local_phase not in valid_phases: return None if history_phase not in valid_phases: return None if not isinstance(before_t, (int, float)): return None if not isinstance(after_t, (int, float)): return None if float(after_t) < float(before_t): return None if not evidence: return None positive_phases = {"defining_motion", "changed_state", "completed"} answer = ( "Yes" if local_phase in positive_phases or history_phase in positive_phases else "No" ) return { "local_phase": local_phase, "history_phase": history_phase, "before_t": float(before_t), "after_t": float(after_t), "evidence": evidence[:1000], "answer": answer, } def normalize_count_audit_response( response: str, question: str, ) -> Optional[str]: """Convert an audit's visible count conclusion to the query format.""" visible = ovo._strip_thinking(response or "").strip() if not visible: return None if RESPONSE_TAG in visible: return visible option_match = re.fullmatch( r"\s*([A-K])\s*[.)::]\s*(\d{1,2})\s*", visible, re.IGNORECASE, ) if option_match: letter = option_match.group(1).upper() count = int(option_match.group(2)) if 0 <= count <= 10 and ord(letter) - ord("A") == count: return f"{RESPONSE_TAG} {letter}. {count}" candidates: List[int] = [] patterns = ( r"\b(?:final\s+)?(?:answer|count|total|running\s+total)\s*" r"(?:is|=|:)\s*(?:option\s+)?(?:[A-K][.)]\s*)?(\d{1,2})\b", r"\bexactly\s+(\d{1,2})\s+(?:distinct\s+)?" r"(?:times?|occurrences?|repetitions?|events?|actions?)\b", r"\b(\d{1,2})\s+(?:distinct\s+)?occurrences?\s+in\s+total\b", ) for pattern in patterns: candidates.extend( int(match) for match in re.findall(pattern, visible, re.I) ) if re.search( r"\b(?:no|zero)\s+(?:valid\s+|distinct\s+)?" r"(?:occurrences?|repetitions?|matching\s+events?)\b", visible, re.I, ): candidates.append(0) candidates = [value for value in candidates if 0 <= value <= 10] if not candidates: return None count = candidates[-1] has_letter_options = bool( re.search(r"(?m)^A\.\s*0\s*$", question or "") and re.search(r"(?m)^K\.\s*10\s*$", question or "") ) if has_letter_options: return f"{RESPONSE_TAG} {chr(ord('A') + count)}. {count}" return f"{RESPONSE_TAG} {count}" def extract_count_audit_value(response: str) -> Optional[int]: """Extract a local 0..10 count from one visual audit segment.""" visible = ovo._strip_thinking(response or "").strip() if RESPONSE_TAG in visible: visible = visible.split(RESPONSE_TAG, 1)[-1].strip() direct = re.fullmatch(r"\s*(\d{1,2})\s*", visible) if direct: value = int(direct.group(1)) return value if 0 <= value <= 10 else None option = re.fullmatch( r"\s*([A-K])\s*[.)::]\s*(\d{1,2})\s*", visible, re.IGNORECASE, ) if option: value = int(option.group(2)) if 0 <= value <= 10 and ord(option.group(1).upper()) - ord("A") == value: return value normalized = normalize_count_audit_response(response, "") if normalized: match = re.search(r"\b(\d{1,2})\s*$", normalized) if match: value = int(match.group(1)) return value if 0 <= value <= 10 else None return None def extract_count_audit_ledger(response: str) -> Optional[List[float]]: """Extract the final JSON timestamp/evidence array from a count audit.""" visible = ovo._strip_thinking(response or "").strip() arrays = re.findall(r"\[[^\[\]]*\]", visible, re.S) if not arrays: return None payload = arrays[-1] try: parsed = json.loads(payload) except Exception: parsed = None values: List[float] = [] if isinstance(parsed, list): for item in parsed: if isinstance(item, (int, float)) and float(item) >= 0: values.append(float(item)) elif isinstance(item, dict): timestamp = item.get("t") evidence = str(item.get("e") or "").strip() if isinstance(timestamp, str): timestamp_match = re.search( r"-?\d+(?:\.\d+)?", timestamp) timestamp = ( float(timestamp_match.group(0)) if timestamp_match else None ) if ( isinstance(timestamp, (int, float)) and float(timestamp) >= 0 and evidence ): values.append(float(timestamp)) else: for token in re.findall(r"-?\d+(?:\.\d+)?", payload): value = float(token) if value >= 0: values.append(value) deduped: List[float] = [] for value in values: if not deduped or abs(value - deduped[-1]) > 0.35: deduped.append(value) return deduped[:10] def split_count_audit_frames( frame_paths: List[str], max_frames: int, ) -> List[List[str]]: """Split a timestamped timeline into contiguous, non-overlapping chunks.""" limit = max(1, int(max_frames)) return [ list(frame_paths[start:start + limit]) for start in range(0, len(frame_paths), limit) ] def merge_count_audit_ledgers( existing: List[float], new_values: List[float], *, tolerance: float = 0.35, ) -> List[float]: """Merge absolute timestamps and remove only near-identical boundaries.""" merged = sorted( float(value) for value in [*existing, *new_values] if isinstance(value, (int, float)) and float(value) >= 0 ) deduped: List[float] = [] for value in merged: if not deduped or abs(value - deduped[-1]) > tolerance: deduped.append(value) return deduped _SPORTS_RELEASE_CYCLE_RULES: Tuple[Tuple[str, str], ...] = ( ( r"\bshot\s*put\b", "One occurrence is one athlete's complete shot-put attempt: a visible " "set/glide/turn and arm drive culminates in the SHOT visibly leaving " "the hand. Count ONLY that release. The same attempt's preparation, " "follow-through, shot flight/landing, measurement, celebration, or " "replay is not another occurrence. A new occurrence requires a new " "athlete/setup and a new shot release.", ), ( r"\bjavelin\s+throw\b", "One occurrence is one complete javelin attempt: run-up/plant and arm " "throw culminate in the JAVELIN visibly leaving the hand. Count ONLY " "that release. The same attempt's run-up frames, follow-through, " "javelin flight/landing, measurement, celebration, or replay is not " "another occurrence. A new occurrence requires a new setup/run-up and " "a new javelin release.", ), ( r"\bcricket\s+bowling\b", "One occurrence is one bowler delivery: bowler run-up and delivery " "stride culminate in the BALL visibly leaving the BOWLER'S hand. " "Count ONLY that bowler release. Ball bounce, batsman hit/swing, " "fielding, celebration, follow-through, and replay are not bowling " "occurrences. A new occurrence requires a new bowler run-up/setup and " "a new bowler release.", ), ) def count_audit_sports_cycle_version(question: str) -> int: """Return the active opt-in sports schema version for this query.""" version = 0 if os.environ.get("STREAMO_ENABLE_REC_SPORTS_CYCLE_V2", "0") == "1": version = 2 if os.environ.get("STREAMO_ENABLE_REC_SPORTS_CYCLE_V3", "0") == "1": version = 3 if not version: return 0 if not any( re.search(pattern, question or "", re.IGNORECASE) for pattern, _ in _SPORTS_RELEASE_CYCLE_RULES ): return 0 return version def count_audit_uses_sports_cycle_schema(question: str) -> bool: """Return whether an opt-in sports release-cycle schema applies.""" if not count_audit_sports_cycle_version(question): return False return any( re.search(pattern, question or "", re.IGNORECASE) for pattern, _ in _SPORTS_RELEASE_CYCLE_RULES ) def extract_sports_cycle_ledger( response: str, target_activity: str, *, require_grounded: bool = False, ) -> Optional[List[float]]: """Parse and validate one segment's structured sports release cycles.""" visible = ovo._strip_thinking(response or "").strip() arrays = re.findall(r"\[[^\[\]]*\]", visible, re.S) if not arrays: return None try: payload = json.loads(arrays[-1]) except (TypeError, ValueError, json.JSONDecodeError): return None if not isinstance(payload, list): return None target = str(target_activity or "").lower() if re.search(r"\bshot\s*put\b", target): object_re = re.compile(r"\b(?:shot|shotput|ball)\b", re.I) elif re.search(r"\bjavelin\b", target): object_re = re.compile(r"\bjavelin\b", re.I) elif re.search(r"\bcricket\s+bowling\b", target): object_re = re.compile(r"\b(?:ball|bowl(?:er|ing)?|delivery)\b", re.I) else: return None release_re = re.compile( r"\b(?:release[sd]?|leaves?|left|letting\s+go|lets?\s+go)\b", re.I, ) placeholder_re = re.compile( r"(?:visible setup/run-up before release|" r"visible target implement/ball leaves hand|" r"visible follow-through/flight after release|" r"specific appearance/location|" r"specific object and its pre-release location|" r"specific pre-release body/object state|" r"specific hand-object separation|" r"specific post-release spatial state|" r"concrete pixels in this cycle)", re.I, ) actor_detail_re = re.compile( r"\b(?:left|right|foreground|background|near|far|red|white|black|" r"blue|green|yellow|orange|dark|light|striped|number|jersey|shirt|" r"jacket|uniform|circle|crease|lane|field)\b", re.I, ) implement_detail_re = re.compile( r"\b(?:left|right|hand|hands|neck|shoulder|behind|beside|near|above|" r"below|chest|head|red|white|black|blue|green|yellow|orange|dark|" r"light)\b", re.I, ) timestamps: List[float] = [] signatures: Set[Tuple[str, ...]] = set() for index, item in enumerate(payload): if not isinstance(item, dict): return None cycle = item.get("cycle") timestamp = item.get("t") if isinstance(timestamp, str): match = re.search(r"-?\d+(?:\.\d+)?", timestamp) timestamp = float(match.group(0)) if match else None before = " ".join(str(item.get("before") or "").split()) release = " ".join(str(item.get("release") or "").split()) after = " ".join(str(item.get("after") or "").split()) actor = " ".join(str(item.get("actor") or "").split()) implement = " ".join(str(item.get("implement") or "").split()) if cycle != index + 1: return None if not isinstance(timestamp, (int, float)) or float(timestamp) < 0: return None if not before or not release or not after: return None if not object_re.search(release) or not release_re.search(release): return None if require_grounded: fields = (actor, implement, before, release, after) if any(not field for field in fields): return None if placeholder_re.search(" ".join(fields)): return None if re.fullmatch( r"(?:the\s+)?(?:athlete|person|someone|actor|bowler)", actor, re.I, ): return None if not actor_detail_re.search(actor): return None if not object_re.search(implement): return None if not implement_detail_re.search(implement): return None normalized = tuple( re.sub(r"[^a-z0-9]+", " ", field.lower()).strip() for field in fields ) if len({normalized[2], normalized[3], normalized[4]}) < 3: return None if normalized in signatures: return None signatures.add(normalized) value = float(timestamp) if timestamps and value <= timestamps[-1]: return None timestamps.append(value) return timestamps[:10] _TIMESTAMP_LEDGER_CYCLE_RE = re.compile( r"\b(?:lifting\b.*\bplacing\b.*\bback\b|tearing\b|opening\b|" r"covering\b|turning\b.*\boff\b|tying\b|buttoning\b|folding\b)", re.IGNORECASE, ) def count_audit_uses_evidence_ledger(question: str) -> bool: """Choose an audit schema from the target action's query semantics.""" return not bool(_TIMESTAMP_LEDGER_CYCLE_RE.search(question or "")) def count_audit_cycle_hint(question: str) -> str: """Return an observable occurrence-cycle definition from query semantics.""" text = question or "" if count_audit_uses_sports_cycle_schema(text): for pattern, rule in _SPORTS_RELEASE_CYCLE_RULES: if re.search(pattern, text, re.IGNORECASE): return rule rules = ( ( r"\bclean\s+and\s+jerk\b", "One occurrence is one complete Olympic clean-and-jerk cycle " "performed with a barbell or weight implement: the implement is " "cleaned from below to the shoulders/front rack, then jerked to " "a stable overhead lockout. Record ONLY the overhead lockout " "completion, once per full cycle. A ball/shot throw, generic " "lifting without the clean-to-overhead sequence, preparation, " "static overhead holding, follow-through, or dropping/putting " "down alone is NOT a clean-and-jerk. Another occurrence requires " "a reset followed by a new complete clean-and-jerk cycle.", ), ( r"\blifting\b.*\bplacing\b.*\bback\b", "One occurrence is the entire cycle: object leaves its support " "surface, remains lifted, then is placed back onto a support " "surface. Record ONLY the final place-back completion time. Do " "not record lift onset, mid-air states, or the same set-down " "across multiple frames.", ), ( r"\btaking\b.*\bout of\b", "One occurrence is one independently handled object completing a " "container/interior-to-outside transfer. Count separate objects " "or separate extraction cycles; continuous holding outside is " "not new.", ), ( r"\bclosing\b", "One occurrence is an open-to-closed state transition completed " "for one object. Record the completion when the lid/door/flap " "reaches closed. Mere touching, moving toward, or holding closed " "is not new.", ), ( r"\bcovering\b", "One occurrence completes when a cover object is placed onto a " "target and the target becomes covered. Adjustment while already " "covered is not new; count another cycle only after reset/uncover " "and re-cover.", ), ( r"\btearing\b", "Each distinct tear stroke that visibly creates a new material " "separation, rip, or newly detached piece is one occurrence. " "Continuous pulling during the same tear is one; a new grip/reset " "followed by another separation is a new occurrence.", ), ( r"\bhitting\b.*\b(?:against|with)\b", "Each distinct impact/contact of one object, hand, or tool against " "another is one occurrence. Rapid repeated taps/strikes count " "separately when contact is followed by separation/reset. A swing " "without contact, holding, or setup is not an occurrence.", ), ( r"\bshowing\b.*\bcamera\b", "One occurrence is a deliberate present-to-camera cycle: object " "is brought toward/faced to the camera, held visibly, then " "withdrawn or reset before another presentation. Record one " "representative completion/peak time per cycle; continuous " "holding is not repeated.", ), ) for pattern, rule in rules: if re.search(pattern, text, re.IGNORECASE): return rule return ( "Define one occurrence by a decisive completion/result and a reset " "before the next occurrence." ) def count_audit_target(question: str) -> str: """Extract the visible activity phrase from a repetition-count query.""" match = re.search( r"activity\s+['\"](.+?)['\"]\s+occur", question or "", re.IGNORECASE, ) if match: return match.group(1).strip() return str(question or "").splitlines()[0].strip() def build_count_audit_prompt( *, target_activity: str, cycle_hint: str, use_evidence_ledger: bool, sports_cycle_version: int = 0, segmented: bool = False, segment_index: int = 0, segment_count: int = 1, segment_start_t: float = 0.0, segment_end_t: float = 0.0, ) -> str: """Build the production count-audit prompt for one visible segment.""" segment_instruction = "" audit_scope = "entire " if segmented: audit_scope = "" segment_instruction = ( f"This is segment {segment_index + 1}/{segment_count}, covering " f"absolute timestamps approximately {segment_start_t:.2f}s to " f"{segment_end_t:.2f}s. Count an occurrence in this response " "ONLY when its decisive completion timestamp falls inside this " "segment. Use the ABSOLUTE timestamp printed on the frame, not " "time relative to the segment. " ) audit_rules = ( "