OUTPUT #734 | 22205 karakter | 490 satır
========== YENİ KOMUT | 17.09.2026 17:04:15 ==========
========== YENİ KOMUT | 17.09.2026 17:04:15 ==========
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.
468- Final production must go through the canonical shorts_fast_pipeline.
469- """
470- if tool_name == "terminal":
471- cmd = str(args.get("command") or "")
472- low = cmd.lower()
473- if "shorts_fast_pipeline.py" in low:
474- return False
475- if "google-tts" in low or "voice_raw_google" in low:
--
530- obj, _ = dec.raw_decode(text[i:])
531- if isinstance(obj, dict):
532- return obj
533- except Exception:
534- continue
535- return None
536-
537-
538-def _normalized_proof_token(value: Any) -> str:
539- """Accept the harmless exact-prefix variation repeatedly emitted by Vision."""
540- token = str(value or "").strip()
541- prefix = "HERMES_PROOF_GATE::"
542- return token[len(prefix):] if token.startswith(prefix) else token
543-
544-
545-def _proof_verdict_accepted(verdict: Any, token: str) -> bool:
546- """Accept proof only when every selected footage role is unambiguous."""
547- return bool(
548- isinstance(verdict, dict)
549- and verdict.get("verdict") == "PASS"
550: and verdict.get("verifier") == "vision_analyze"
551- and verdict.get("primary_visual_proof") is True
552- and verdict.get("hook_visual_goal") is True
553- and verdict.get("proof_visible_and_unmistakable") is True
554- and verdict.get("casting_subjects_clear") is True
555- and _normalized_proof_token(verdict.get("proof_token")) == token
556- )
557-
558-
559-def _on_pre_tool_call(
560- tool_name: str = "",
561- args: Any = None,
562- session_id: str = "",
563- turn_id: str = "",
564- **_: Any,
565-) -> Optional[dict]:
566- active, alternate_method, explicit_system_change = _production_context(session_id)
567- if not active:
568- return None
569-
570- args = args if isinstance(args, dict) else {}
--
608-
609- if not alternate_method and not explicit_system_change and _direct_render_bypass(tool_name, args):
610- return _block(
611- "Shorts production guard: direct TTS/final assembly outside the canonical pipeline "
612- "is blocked during normal production. Use shorts_fast_pipeline.py. If proof Vision "
613- "cannot issue a valid receipt, stop with QA_UNAVAILABLE; do not build a fallback final."
614- )
615-
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."
--
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(
705- "Shorts production guard: do not modify skills during a production job. "
706- "Use the current canonical pipeline and finish the artifact."
707- )
708-
709- if tool_name in {"write_file", "patch"}:
710- path = str(
711- args.get("path")
712- or args.get("file_path")
713- or ""
714- )
715- if path.endswith("proof_receipt.json") or path.endswith("proof_pending.json"):
716- return _block(
717- "Shorts production guard: proof state is runtime-owned. "
718- "Do not write or patch proof receipts/pending attestations manually; "
719: "only a real vision_analyze PASS may issue the receipt."
720- )
721- if any(path.startswith(prefix) for prefix in _PROTECTED_PREFIXES):
722- return _block(
723- "Shorts production guard: production-time mutation of Hermes core/tools/skills "
724- "is blocked. Use the installed production stack as-is and finish the video."
725- )
726-
727- if tool_name == "terminal":
728- cmd = str(args.get("command") or "")
729- low = cmd.lower()
730- if "proof_receipt.json" in low or "proof_pending.json" in low:
731- return _block(
732- "Shorts production guard: proof state is runtime-owned. "
733- "Do not create, edit, copy, delete, or inspect proof receipt/pending files "
734- "through terminal commands during production."
735- )
736- if (
737- _INSTALL_RE.search(cmd)
738- or "local-vlm" in low
739- or ("transformers" in low and "install" in low)
740- or ("torch" in low and "install" in low)
741- ):
742- 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
814- else:
815- accepted = (
816- verdict.get("verdict") == "PASS"
817: and verdict.get("verifier") == "vision_analyze"
818- and verdict.get("post_repair_semantic_verified") is True
819- and _normalized_proof_token(verdict.get("repair_token")) == token
820- )
821- failed = verdict.get("verdict") == "FAIL"
822- state["state"] = "verified" if accepted else ("failed" if failed else "unverified")
823- state["verification_pending"] = False
824- state["claim_fixed_allowed"] = accepted
825- state["verification"] = {
826: "verifier": "vision_analyze",
827- "verdict": verdict.get("verdict"),
828- "artifact_sha256": state.get("artifact_sha256"),
829- "contact_sheet_sha256": state.get("contact_sheet_sha256"),
830- "session_id": session_id,
831- "turn_id": turn_id,
832- "tool_call_id": tool_call_id,
833- "accepted_at": time.time(),
834- "raw": verdict,
835- "vision_route": outer.get("vision_route") if isinstance(outer.get("vision_route"), dict) else None,
836- }
837- try:
838- tmp = state_path.with_suffix(state_path.suffix + ".tmp")
839- tmp.write_text(json.dumps(state, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
840- tmp.replace(state_path)
841- except Exception:
842- return
843-
844-
845-def _on_post_tool_call(
846- tool_name: str = "",
847- args: Any = None,
848- result: Any = None,
849- session_id: str = "",
850- turn_id: str = "",
851- tool_call_id: str = "",
852- status: str = "",
853- **_: Any,
854-) -> None:
855: if tool_name != "vision_analyze":
856- return
857-
858- active, _, _ = _production_context(session_id)
859- if not active:
860- return
861- args = args if isinstance(args, dict) else {}
862- image_url = str(args.get("image_url") or "")
863- question = str(args.get("question") or "")
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:
974- return
975-
976-def _on_subagent_start(
977- parent_session_id: str = "",
978- child_session_id: str = "",
979- child_goal: str = "",
980- **_: Any,
981-) -> None:
root@213-238-170-219:~#