========== YENİ KOMUT | 17.09.2026 20:09:23 ==========
========== YENİ KOMUT | 17.09.2026 20:09:23 ==========
### L545-L558
def _proof_verdict_accepted(verdict: Any, token: str) -> bool:
"""Accept proof only when every selected footage role is unambiguous."""
return bool(
isinstance(verdict, dict)
and verdict.get("verdict") == "PASS"
and verdict.get("verifier") == "vision_analyze"
and verdict.get("primary_visual_proof") is True
and verdict.get("hook_visual_goal") is True
and verdict.get("proof_visible_and_unmistakable") is True
and verdict.get("casting_subjects_clear") is True
and _normalized_proof_token(verdict.get("proof_token")) == token
)
### L559-L754
def _on_pre_tool_call(
tool_name: str = "",
args: Any = None,
session_id: str = "",
turn_id: str = "",
**_: Any,
) -> Optional[dict]:
active, alternate_method, explicit_system_change = _production_context(session_id)
if not active:
return None
args = args if isinstance(args, dict) else {}
# Normal Shorts may delegate only the bounded Astra creative-director plan/re-plan.
# All other production delegation remains blocked.
if tool_name == "delegate_task" and not alternate_method and not explicit_system_change:
tasks = args.get("tasks") if isinstance(args.get("tasks"), list) else []
goals = [
str(t.get("goal") or "")
for t in tasks
if isinstance(t, dict)
]
astra_planner = (
len(goals) == 1
and goals[0].startswith("HERMES_CREATIVE_DIRECTOR")
)
if not astra_planner:
return _block(
"Shorts production guard: normal production delegation is restricted to the "
"bounded Astra creative director. Use one delegate_task whose goal starts with "
"HERMES_CREATIVE_DIRECTOR; all other delegation remains blocked."
)
typed_tasks = [dict(tasks[0])]
typed_tasks[0]["goal"] = (
str(typed_tasks[0].get("goal") or "")
+ "\nKeep the creative contract compact and concise. Use short phrases/sentences; "
"include only information needed by the production executor."
)
typed_tasks[0]["output_schema"] = _CREATIVE_CONTRACT_SCHEMA
return {"action": "modify", "args": {"tasks": typed_tasks}}
if not alternate_method and not explicit_system_change:
rediscovery_block = _canonical_rediscovery_block(tool_name, args, session_id)
if rediscovery_block:
return rediscovery_block
freeze_block = _proof_freeze_block(tool_name, args)
if freeze_block:
return freeze_block
if not alternate_method and not explicit_system_change and _direct_render_bypass(tool_name, args):
return _block(
"Shorts production guard: direct TTS/final assembly outside the canonical pipeline "
"is blocked during normal production. Use shorts_fast_pipeline.py. If proof Vision "
"cannot issue a valid receipt, stop with QA_UNAVAILABLE; do not build a fallback final."
)
# Explicitly requested alternate/code-driven workflows remain available.
if tool_name == "skill_view" and not alternate_method:
name = str(args.get("name") or "")
if name and name != "shorts-core":
return _block(
"Shorts production guard: shorts-core is the sole production authority "
"for this footage-based Shorts/Reels turn. Do not load another production, "
"debug, installation, or QA skill. Continue with shorts-core and the canonical pipeline."
)
# Vision policy: do not impose arbitrary call counts.
# The agent may re-evaluate when the evidence or question genuinely changes.
# Only exact no-progress repetition of the same visual evidence + question is blocked.
if tool_name in {"browser_vision", "computer_use"} and not alternate_method:
return _block(
"Shorts production guard: this QA fallback is disabled during normal production. "
"Use vision_analyze as the primary visual QA path; if Vision is unavailable after "
"the permitted retry, continue with controlled QA_UNAVAILABLE rather than inventing a fallback."
)
if tool_name == "vision_analyze":
image_url = str(args.get("image_url") or "")
question = str(args.get("question") or "").strip()
raw = image_url[7:] if image_url.startswith("file://") else image_url
image_path = Path(raw).expanduser()
# ASSET-001 diagnostic:
# Pre-casting source/range/asset Vision inspection is intentionally OPEN.
# No artificial Vision-call cap and no mandatory batching in this diagnostic.
# Use visual inspection only when it helps choose footage that actually fulfills
# the creative contract. Existing duplicate/no-progress and other guards remain active.
if image_path.name == "proof_contact_sheet.jpg":
pending_path = image_path.parent / "proof_pending.json"
current_fingerprint = None
pending = {}
try:
if pending_path.exists():
pending = json.loads(pending_path.read_text(encoding="utf-8"))
current_fingerprint = pending.get("fingerprint")
except Exception:
pending = {}
proof_state = _read_proof_attempt_state(image_path, pending)
if (
int(proof_state.get("failed_attempts") or 0) >= 2
and current_fingerprint
and proof_state.get("last_fingerprint") == current_fingerprint
):
return _block(
"Shorts production guard: PROOF_STRATEGY_EXHAUSTED — this proof strategy "
"already failed initial review plus one repair. Do not retry Vision on the "
"same accepted candidates/casting. Materially change the proof footage or "
"casting, rerun proof-check, and continue the production autonomously."
)
# Content identity matters more than filename: a repaired final_contact_sheet
# at the same path is new evidence and must be allowed to be judged again.
try:
image_identity = _sha_file(image_path.resolve())
except Exception:
image_identity = str(image_path)
normalized_question = re.sub(r"\s+", " ", question).strip()
sig_raw = json.dumps(
{
"session": session_id,
"image": image_identity,
"question": normalized_question,
},
ensure_ascii=False,
sort_keys=True,
)
signature = hashlib.sha256(sig_raw.encode("utf-8")).hexdigest()
with _LOCK:
if signature in _VISION_SEEN:
return _block(
"Shorts production guard: this exact visual evidence has already been "
"asked the same question. Reuse the existing result. If evidence was repaired "
"or the verification question genuinely needs correction, change the evidence "
"or question rather than repeating the same call."
)
return None
# During a production turn the agent may USE the installed toolchain, but may not redesign it.
if not explicit_system_change:
if tool_name == "skill_manage":
return _block(
"Shorts production guard: do not modify skills during a production job. "
"Use the current canonical pipeline and finish the artifact."
)
if tool_name in {"write_file", "patch"}:
path = str(
args.get("path")
or args.get("file_path")
or ""
)
if path.endswith("proof_receipt.json") or path.endswith("proof_pending.json"):
return _block(
"Shorts production guard: proof state is runtime-owned. "
"Do not write or patch proof receipts/pending attestations manually; "
"only a real vision_analyze PASS may issue the receipt."
)
if any(path.startswith(prefix) for prefix in _PROTECTED_PREFIXES):
return _block(
"Shorts production guard: production-time mutation of Hermes core/tools/skills "
"is blocked. Use the installed production stack as-is and finish the video."
)
if tool_name == "terminal":
cmd = str(args.get("command") or "")
low = cmd.lower()
if "proof_receipt.json" in low or "proof_pending.json" in low:
return _block(
"Shorts production guard: proof state is runtime-owned. "
"Do not create, edit, copy, delete, or inspect proof receipt/pending files "
"through terminal commands during production."
)
if (
_INSTALL_RE.search(cmd)
or "local-vlm" in low
or ("transformers" in low and "install" in low)
or ("torch" in low and "install" in low)
):
return _block(
"Shorts production guard: installing packages/models or creating a local VLM "
"during video production is blocked. Do not build a new QA/tool stack; continue "
"with the canonical installed pipeline."
)
if _PROTECTED_MUTATION_RE.search(cmd):
return _block(
"Shorts production guard: production-time mutation of the Hermes runtime/toolchain "
"is blocked. Use existing tools without modifying them."
)
return None
### L778-L844
def _update_repair_state_from_vision(
image_path: Path,
question: str,
result: Any,
session_id: str,
turn_id: str,
tool_call_id: str,
) -> None:
if image_path.name != "final_contact_sheet.jpg":
return
state_path = image_path.parent / "repair_state.json"
if not state_path.exists():
return
try:
state = json.loads(state_path.read_text(encoding="utf-8"))
except Exception:
return
if state.get("state") not in {"repair_attempted", "verification_pending"}:
return
token = str(state.get("repair_token") or "")
if not token or f"HERMES_REPAIR_VERIFY::{token}" not in question:
return
try:
if _sha_file(image_path) != state.get("contact_sheet_sha256"):
return
artifact = Path(str(state.get("artifact") or "")).expanduser().resolve()
if not artifact.exists() or _sha_file(artifact) != state.get("artifact_sha256"):
return
except Exception:
return
outer = _json_object(result)
verdict = _json_object(outer.get("analysis")) if outer and outer.get("success") is True else None
if not verdict:
state["state"] = "unverified"
state["verification_pending"] = False
state["claim_fixed_allowed"] = False
else:
accepted = (
verdict.get("verdict") == "PASS"
and verdict.get("verifier") == "vision_analyze"
and verdict.get("post_repair_semantic_verified") is True
and _normalized_proof_token(verdict.get("repair_token")) == token
)
failed = verdict.get("verdict") == "FAIL"
state["state"] = "verified" if accepted else ("failed" if failed else "unverified")
state["verification_pending"] = False
state["claim_fixed_allowed"] = accepted
state["verification"] = {
"verifier": "vision_analyze",
"verdict": verdict.get("verdict"),
"artifact_sha256": state.get("artifact_sha256"),
"contact_sheet_sha256": state.get("contact_sheet_sha256"),
"session_id": session_id,
"turn_id": turn_id,
"tool_call_id": tool_call_id,
"accepted_at": time.time(),
"raw": verdict,
"vision_route": outer.get("vision_route") if isinstance(outer.get("vision_route"), dict) else None,
}
try:
tmp = state_path.with_suffix(state_path.suffix + ".tmp")
tmp.write_text(json.dumps(state, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
tmp.replace(state_path)
except Exception:
return
### L845-L975
def _on_post_tool_call(
tool_name: str = "",
args: Any = None,
result: Any = None,
session_id: str = "",
turn_id: str = "",
tool_call_id: str = "",
status: str = "",
**_: Any,
) -> None:
if tool_name != "vision_analyze":
return
active, _, _ = _production_context(session_id)
if not active:
return
args = args if isinstance(args, dict) else {}
image_url = str(args.get("image_url") or "")
question = str(args.get("question") or "")
if not image_url:
return
# A provider failure is not reviewed evidence. Mark only completed Vision
# calls as seen so the provider's explicit retry can reuse the exact sheet
# and question instead of forcing a meaningless evidence mutation.
raw_identity = image_url[7:] if image_url.startswith("file://") else image_url
try:
image_identity = _sha_file(Path(raw_identity).expanduser().resolve())
except Exception:
image_identity = str(Path(raw_identity).expanduser())
normalized_question = re.sub(r"\s+", " ", question).strip()
signature = hashlib.sha256(json.dumps(
{"session": session_id, "image": image_identity, "question": normalized_question},
ensure_ascii=False, sort_keys=True,
).encode("utf-8")).hexdigest()
outer = _json_object(result)
if status and status != "ok":
return
if not outer or outer.get("success") is not True:
return
with _LOCK:
_VISION_SEEN.add(signature)
raw_path = image_url[7:] if image_url.startswith("file://") else image_url
try:
image_path = Path(raw_path).expanduser().resolve()
except Exception:
return
contract_question = _expanded_vision_question_for_receipt(image_path, question)
_update_repair_state_from_vision(
image_path, contract_question, result, session_id, turn_id, tool_call_id
)
if image_path.name != "proof_contact_sheet.jpg":
return
pending_path = image_path.parent / "proof_pending.json"
if not pending_path.exists():
return
try:
pending = json.loads(pending_path.read_text(encoding="utf-8"))
except Exception:
return
token = str(pending.get("proof_token") or "")
if not token or f"HERMES_PROOF_GATE::{token}" not in contract_question:
return
try:
expected_sheet = Path(str(pending["contact_sheet"])).resolve()
if expected_sheet != image_path:
return
if _sha_file(image_path) != pending.get("contact_sheet_sha256"):
return
except Exception:
return
outer = _json_object(result)
if not outer or outer.get("success") is not True:
return
verdict = _json_object(outer.get("analysis"))
if not verdict:
return
accepted = _proof_verdict_accepted(verdict, token)
if not accepted:
if verdict.get("verdict") == "FAIL":
state = _read_proof_attempt_state(image_path, pending)
fingerprint = pending.get("fingerprint")
if state.get("last_fingerprint") != fingerprint:
state["failed_attempts"] = 0
state["failed_attempts"] = int(state.get("failed_attempts") or 0) + 1
state["last_verdict"] = "FAIL"
state["last_fingerprint"] = fingerprint
state["updated_at"] = time.time()
_write_proof_attempt_state(image_path, state, pending)
return
state = _read_proof_attempt_state(image_path, pending)
state["accepted"] = True
state["last_verdict"] = "PASS"
state["last_fingerprint"] = pending.get("fingerprint")
state["updated_at"] = time.time()
_write_proof_attempt_state(image_path, state, pending)
try:
receipt = Path(str(pending["receipt"])).expanduser().resolve()
receipt.parent.mkdir(parents=True, exist_ok=True)
obj = {
"pass": True,
"receipt_source": "shorts-production-guard/post_tool_call",
"verifier": "vision_analyze",
"fingerprint": pending["fingerprint"],
"contact_sheet": str(image_path),
"contact_sheet_sha256": pending["contact_sheet_sha256"],
"verdict": verdict,
"vision_route": outer.get("vision_route") if isinstance(outer.get("vision_route"), dict) else None,
"session_id": session_id,
"turn_id": turn_id,
"tool_call_id": tool_call_id,
"accepted_at": time.time(),
}
tmp = receipt.with_suffix(receipt.suffix + ".tmp")
tmp.write_text(json.dumps(obj, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
tmp.replace(receipt)
except Exception:
return
root@213-238-170-219:~#