OUTPUT #722 | 21817 karakter | 495 satır
========== YENİ KOMUT | 17.09.2026 15:53:57 ==========
========== YENİ KOMUT | 17.09.2026 15:53:57 ==========
1-from __future__ import annotations
2-
3:import hashlib
4-import json
5-import re
6-import time
7-import sqlite3
8-import threading
9-from pathlib import Path
10-from typing import Any, Optional
11-
12-_DB = Path.home() / ".hermes" / "state.db"
13-_LOCK = threading.Lock()
14:_VISION_SEEN: set[str] = set()
15-_CANONICAL_READS: set[tuple[str, str]] = set()
16-_CREATIVE_CHILDREN: dict[str, str] = {}
17-
18-_CREATIVE_CONTRACT_DIR = Path.home() / ".hermes" / "cache" / "shorts-production-guard" / "creative-contracts"
19-_CREATIVE_CONTRACT_SCHEMA = {
20- "type": "object",
21- "additionalProperties": False,
22- "properties": {
23- "contract_version": {"type": "integer", "const": 1},
24- "creative_thesis": {"type": "string", "minLength": 1},
25- "visual_strategy": {"type": "string", "minLength": 1},
26- "source_strategy": {"type": "string", "minLength": 1},
--
247- "Shorts production guard: normal production must use the documented interface "
248- f"instead of re-reading {target} to rediscover schema. Run the canonical "
249- "preflight/proof-check/pipeline command. Source and tests remain readable in an "
250- "explicit engineering/system-change task."
251- )
252- _CANONICAL_READS.add(key)
253- return None
254-
255-def _block(message: str) -> dict[str, str]:
256- return {"action": "block", "message": message}
257-
258-
259:def _runtime_proof_fingerprint(data: dict) -> str:
260: """Mirror the canonical proof fingerprint so accepted proof can be frozen at runtime."""
261- creative = data.get("creative_intent") or {}
262- gate = data.get("proof_gate") or {}
263- candidates = gate.get("candidates") or []
264-
265- normalized = []
266- for c in candidates:
267- if not isinstance(c, dict):
268- normalized.append(c)
269- continue
270- item = dict(c)
271- src = item.get("src")
272- if isinstance(src, str) and src.strip():
--
328- "creative_intent": {
329- "promise": creative.get("promise"),
330- "primary_visual_proof": creative.get("primary_visual_proof"),
331- "hook_visual_goal": creative.get("hook_visual_goal"),
332- "payoff": creative.get("payoff"),
333- },
334- "proof_gate": {
335- "mode": gate.get("mode"),
336- "candidates": normalized,
337- },
338- "selected_casting": selected_casting,
339- }
340: return hashlib.sha256(
341- json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8")
342- ).hexdigest()
343-
344-
345-def _proof_freeze_block(tool_name: str, args: dict) -> Optional[dict]:
346- """After a valid proof PASS, prevent edits that invalidate the accepted proof."""
347- if tool_name not in {"patch", "write_file", "execute_code"}:
348- return None
349-
350- if tool_name == "execute_code":
351- code = str(args.get("code") or "")
352- low = code.lower()
353- if "manifest.json" in low and any(
354- marker in low
355- for marker in ("write_file(", "write_text(", ".write(", "json.dump(", "open(")
356- ):
357- return _block(
358- "Shorts production guard: PROOF_FREEZE — do not mutate manifest.json through "
359- "execute_code after an accepted proof. Use patch/write_file instead so the runtime "
360: "can compare the prospective proof fingerprint and allow proof-irrelevant edits "
361- "while blocking changes to accepted creative proof/casting."
362- )
363- return None
364-
365- raw_path = str(args.get("path") or args.get("file_path") or "")
366- if not raw_path:
367- return None
368- try:
369- path = Path(raw_path).expanduser().resolve()
370- except Exception:
371- return None
372- if path.name != "manifest.json" or not path.exists():
--
379- receipt_value = gate.get("receipt")
380- receipt = (
381- Path(str(receipt_value)).expanduser().resolve()
382- if receipt_value
383- else (path.parent / "proof_gate" / "proof_receipt.json").resolve()
384- )
385- if not receipt.exists():
386- return None
387- accepted = json.loads(receipt.read_text(encoding="utf-8"))
388- if accepted.get("pass") is not True:
389- return None
390-
391: current_fp = _runtime_proof_fingerprint(current)
392: if accepted.get("fingerprint") != current_fp:
393- # Already stale: do not create a new policy trap here.
394- return None
395-
396- if tool_name == "write_file":
397- prospective_text = args.get("content")
398- if not isinstance(prospective_text, str):
399- return None
400- else:
401- old = args.get("old_string")
402- new = args.get("new_string")
403- if not isinstance(old, str) or not isinstance(new, str) or old not in current_text:
404- return None
405- count = current_text.count(old)
406- if count > 1 and args.get("replace_all") is not True:
407- return None
408- prospective_text = (
409- current_text.replace(old, new)
410- if args.get("replace_all") is True
411- else current_text.replace(old, new, 1)
412- )
413-
414- prospective = json.loads(prospective_text)
415: if _runtime_proof_fingerprint(prospective) != current_fp:
416- return _block(
417- "Shorts production guard: PROOF_FREEZE — accepted proof is locked. "
418- "Do not change creative_intent, proof candidates, or selected footage/casting "
419- "after Vision PASS. Continue with TTS/render/final QA using the accepted plan. "
420- "A genuine proof failure must be handled before acceptance, not by reopening "
421- "the proof loop after PASS."
422- )
423- except Exception:
424- return None
425- return None
426-
427-
428:def _proof_attempt_state_path(image_path: Path, pending: Optional[dict] = None) -> Path:
429- """Use one attempt ledger per canonical proof receipt, not per proofN workdir."""
430- if isinstance(pending, dict):
431- receipt = pending.get("receipt")
432- if isinstance(receipt, str) and receipt.strip():
433- try:
434- return Path(receipt).expanduser().resolve().parent / "proof_attempt_state.json"
435- except Exception:
436- pass
437: return image_path.parent / "proof_attempt_state.json"
438-
439-
440:def _read_proof_attempt_state(image_path: Path, pending: Optional[dict] = None) -> dict:
441: path = _proof_attempt_state_path(image_path, pending)
442- try:
443- if path.exists():
444- value = json.loads(path.read_text(encoding="utf-8"))
445- return value if isinstance(value, dict) else {}
446- except Exception:
447- pass
448- return {}
449-
450-
451-def _write_proof_attempt_state(
452: image_path: Path, state: dict, pending: Optional[dict] = None
453-) -> None:
454- try:
455: path = _proof_attempt_state_path(image_path, pending)
456- path.parent.mkdir(parents=True, exist_ok=True)
457- tmp = path.with_suffix(path.suffix + ".tmp")
458- tmp.write_text(json.dumps(state, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
459- tmp.replace(path)
460- except Exception:
461- pass
462-
463-
464-def _direct_render_bypass(tool_name: str, args: dict) -> bool:
465- """Block manual TTS/final assembly during normal Shorts production.
466-
467- Asset inspection, proof generation and deterministic QA remain available.
--
496- and any(x in low for x in ("subprocess", "terminal(", "os.system", "run("))
497- ):
498- return True
499- if (
500- "ffmpeg" in low
501- and any(x in low for x in ("final.mp4", "visual_master.mp4"))
502- and any(x in low for x in ("subprocess", "terminal(", "os.system", "run("))
503- ):
504- return True
505- return False
506-
507-def _sha_file(path: Path) -> str:
508: h = hashlib.sha256()
509- with path.open("rb") as f:
510- for chunk in iter(lambda: f.read(1024 * 1024), b""):
511- h.update(chunk)
512- return h.hexdigest()
513-
514-def _json_object(value: Any) -> Optional[dict]:
515- if isinstance(value, dict):
516- return value
517- if not isinstance(value, str):
518- return None
519- text = value.strip()
520- try:
--
616- # Explicitly requested alternate/code-driven workflows remain available.
617- if tool_name == "skill_view" and not alternate_method:
618- name = str(args.get("name") or "")
619- if name and name != "shorts-core":
620- return _block(
621- "Shorts production guard: shorts-core is the sole production authority "
622- "for this footage-based Shorts/Reels turn. Do not load another production, "
623- "debug, installation, or QA skill. Continue with shorts-core and the canonical pipeline."
624- )
625-
626- # Vision policy: do not impose arbitrary call counts.
627- # The agent may re-evaluate when the evidence or question genuinely changes.
628: # Only exact no-progress repetition of the same visual evidence + question is blocked.
629- if tool_name in {"browser_vision", "computer_use"} and not alternate_method:
630- return _block(
631- "Shorts production guard: this QA fallback is disabled during normal production. "
632- "Use vision_analyze as the primary visual QA path; if Vision is unavailable after "
633- "the permitted retry, continue with controlled QA_UNAVAILABLE rather than inventing a fallback."
634- )
635-
636- if tool_name == "vision_analyze":
637- image_url = str(args.get("image_url") or "")
638- question = str(args.get("question") or "").strip()
639- raw = image_url[7:] if image_url.startswith("file://") else image_url
640: image_path = Path(raw).expanduser()
641-
642- # ASSET-001 diagnostic:
643- # Pre-casting source/range/asset Vision inspection is intentionally OPEN.
644- # No artificial Vision-call cap and no mandatory batching in this diagnostic.
645- # Use visual inspection only when it helps choose footage that actually fulfills
646: # the creative contract. Existing duplicate/no-progress and other guards remain active.
647-
648: if image_path.name == "proof_contact_sheet.jpg":
649: pending_path = image_path.parent / "proof_pending.json"
650: current_fingerprint = None
651- pending = {}
652- try:
653- if pending_path.exists():
654- pending = json.loads(pending_path.read_text(encoding="utf-8"))
655: current_fingerprint = pending.get("fingerprint")
656- except Exception:
657- pending = {}
658: proof_state = _read_proof_attempt_state(image_path, pending)
659-
660- if (
661- int(proof_state.get("failed_attempts") or 0) >= 2
662: and current_fingerprint
663: and proof_state.get("last_fingerprint") == current_fingerprint
664- ):
665- return _block(
666: "Shorts production guard: PROOF_STRATEGY_EXHAUSTED — this proof strategy "
667- "already failed initial review plus one repair. Do not retry Vision on the "
668- "same accepted candidates/casting. Materially change the proof footage or "
669- "casting, rerun proof-check, and continue the production autonomously."
670- )
671-
672- # Content identity matters more than filename: a repaired final_contact_sheet
673- # at the same path is new evidence and must be allowed to be judged again.
674- try:
675: image_identity = _sha_file(image_path.resolve())
676- except Exception:
677: image_identity = str(image_path)
678-
679- normalized_question = re.sub(r"\s+", " ", question).strip()
680- sig_raw = json.dumps(
681- {
682- "session": session_id,
683- "image": image_identity,
684- "question": normalized_question,
685- },
686- ensure_ascii=False,
687- sort_keys=True,
688- )
689: signature = hashlib.sha256(sig_raw.encode("utf-8")).hexdigest()
690-
691- with _LOCK:
692: if signature in _VISION_SEEN:
693- return _block(
694- "Shorts production guard: this exact visual evidence has already been "
695- "asked the same question. Reuse the existing result. If evidence was repaired "
696- "or the verification question genuinely needs correction, change the evidence "
697- "or question rather than repeating the same call."
698- )
699- return None
700-
701- # During a production turn the agent may USE the installed toolchain, but may not redesign it.
702- if not explicit_system_change:
703- if tool_name == "skill_manage":
704- return _block(
--
743- "Shorts production guard: installing packages/models or creating a local VLM "
744- "during video production is blocked. Do not build a new QA/tool stack; continue "
745- "with the canonical installed pipeline."
746- )
747- if _PROTECTED_MUTATION_RE.search(cmd):
748- return _block(
749- "Shorts production guard: production-time mutation of the Hermes runtime/toolchain "
750- "is blocked. Use existing tools without modifying them."
751- )
752-
753- return None
754-
755:def _expanded_vision_question_for_receipt(image_path: Path, question: str) -> str:
756- """Resolve only canonical sibling QA prompts for receipt token checks."""
757- prefix = "@file:"
758- text = str(question or "")
759- if not text.startswith(prefix):
760- return text
761- prompt = Path(text[len(prefix):].strip()).expanduser()
762- if not prompt.is_absolute():
763- return text
764- try:
765- prompt = prompt.resolve()
766: if prompt.parent != image_path.parent:
767- return text
768- if not prompt.name.endswith("_vision_question.txt"):
769- return text
770- if not prompt.is_file() or prompt.stat().st_size > 128 * 1024:
771- return text
772- expanded = prompt.read_text(encoding="utf-8").strip()
773- return expanded or text
774- except Exception:
775- return text
776-
777-
778-def _update_repair_state_from_vision(
779: image_path: Path,
780- question: str,
781- result: Any,
782- session_id: str,
783- turn_id: str,
784- tool_call_id: str,
785-) -> None:
786: if image_path.name != "final_contact_sheet.jpg":
787- return
788: state_path = image_path.parent / "repair_state.json"
789- if not state_path.exists():
790- return
791- try:
792- state = json.loads(state_path.read_text(encoding="utf-8"))
793- except Exception:
794- return
795- if state.get("state") not in {"repair_attempted", "verification_pending"}:
796- return
797- token = str(state.get("repair_token") or "")
798- if not token or f"HERMES_REPAIR_VERIFY::{token}" not in question:
799- return
800- try:
801: if _sha_file(image_path) != state.get("contact_sheet_sha256"):
802- return
803- artifact = Path(str(state.get("artifact") or "")).expanduser().resolve()
804- if not artifact.exists() or _sha_file(artifact) != state.get("artifact_sha256"):
805- return
806- except Exception:
807- return
808- outer = _json_object(result)
809- verdict = _json_object(outer.get("analysis")) if outer and outer.get("success") is True else None
810- if not verdict:
811- state["state"] = "unverified"
812- state["verification_pending"] = False
813- state["claim_fixed_allowed"] = False
--
864- if not image_url:
865- return
866-
867- # A provider failure is not reviewed evidence. Mark only completed Vision
868- # calls as seen so the provider's explicit retry can reuse the exact sheet
869- # and question instead of forcing a meaningless evidence mutation.
870- raw_identity = image_url[7:] if image_url.startswith("file://") else image_url
871- try:
872- image_identity = _sha_file(Path(raw_identity).expanduser().resolve())
873- except Exception:
874- image_identity = str(Path(raw_identity).expanduser())
875- normalized_question = re.sub(r"\s+", " ", question).strip()
876: signature = hashlib.sha256(json.dumps(
877- {"session": session_id, "image": image_identity, "question": normalized_question},
878- ensure_ascii=False, sort_keys=True,
879- ).encode("utf-8")).hexdigest()
880- outer = _json_object(result)
881- if status and status != "ok":
882- return
883- if not outer or outer.get("success") is not True:
884- return
885- with _LOCK:
886: _VISION_SEEN.add(signature)
887-
888- raw_path = image_url[7:] if image_url.startswith("file://") else image_url
889- try:
890: image_path = Path(raw_path).expanduser().resolve()
891- except Exception:
892- return
893-
894: contract_question = _expanded_vision_question_for_receipt(image_path, question)
895- _update_repair_state_from_vision(
896: image_path, contract_question, result, session_id, turn_id, tool_call_id
897- )
898-
899: if image_path.name != "proof_contact_sheet.jpg":
900- return
901-
902: pending_path = image_path.parent / "proof_pending.json"
903- if not pending_path.exists():
904- return
905-
906- try:
907- pending = json.loads(pending_path.read_text(encoding="utf-8"))
908- except Exception:
909- return
910-
911- token = str(pending.get("proof_token") or "")
912- if not token or f"HERMES_PROOF_GATE::{token}" not in contract_question:
913- return
914-
915- try:
916- expected_sheet = Path(str(pending["contact_sheet"])).resolve()
917: if expected_sheet != image_path:
918- return
919: if _sha_file(image_path) != pending.get("contact_sheet_sha256"):
920- return
921- except Exception:
922- return
923-
924- outer = _json_object(result)
925- if not outer or outer.get("success") is not True:
926- return
927-
928- verdict = _json_object(outer.get("analysis"))
929- if not verdict:
930- return
931-
932- accepted = _proof_verdict_accepted(verdict, token)
933- if not accepted:
934- if verdict.get("verdict") == "FAIL":
935: state = _read_proof_attempt_state(image_path, pending)
936: fingerprint = pending.get("fingerprint")
937: if state.get("last_fingerprint") != fingerprint:
938- state["failed_attempts"] = 0
939- state["failed_attempts"] = int(state.get("failed_attempts") or 0) + 1
940- state["last_verdict"] = "FAIL"
941: state["last_fingerprint"] = fingerprint
942- state["updated_at"] = time.time()
943: _write_proof_attempt_state(image_path, state, pending)
944- return
945-
946: state = _read_proof_attempt_state(image_path, pending)
947- state["accepted"] = True
948- state["last_verdict"] = "PASS"
949: state["last_fingerprint"] = pending.get("fingerprint")
950- state["updated_at"] = time.time()
951: _write_proof_attempt_state(image_path, state, pending)
952-
953- try:
954- receipt = Path(str(pending["receipt"])).expanduser().resolve()
955- receipt.parent.mkdir(parents=True, exist_ok=True)
956- obj = {
957- "pass": True,
958- "receipt_source": "shorts-production-guard/post_tool_call",
959- "verifier": "vision_analyze",
960: "fingerprint": pending["fingerprint"],
961: "contact_sheet": str(image_path),
962- "contact_sheet_sha256": pending["contact_sheet_sha256"],
963- "verdict": verdict,
964- "vision_route": outer.get("vision_route") if isinstance(outer.get("vision_route"), dict) else None,
965- "session_id": session_id,
966- "turn_id": turn_id,
967- "tool_call_id": tool_call_id,
968- "accepted_at": time.time(),
969- }
970- tmp = receipt.with_suffix(receipt.suffix + ".tmp")
971- tmp.write_text(json.dumps(obj, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
972- tmp.replace(receipt)
973- except Exception:
root@213-238-170-219:~#