========== YENİ KOMUT | 17.09.2026 17:32:26 ==========


========== YENİ KOMUT | 17.09.2026 17:32:26 ==========

from __future__ import annotations

import hashlib
import json
import re
import time
import sqlite3
import threading
from pathlib import Path
from typing import Any, Optional

_DB = Path.home() / ".hermes" / "state.db"
_LOCK = threading.Lock()
_VISION_SEEN: set[str] = set()
_CANONICAL_READS: set[tuple[str, str]] = set()
_CREATIVE_CHILDREN: dict[str, str] = {}

_CREATIVE_CONTRACT_DIR = Path.home() / ".hermes" / "cache" / "shorts-production-guard" / "creative-contracts"
_CREATIVE_CONTRACT_SCHEMA = {
    "type": "object",
    "additionalProperties": False,
    "properties": {
        "contract_version": {"type": "integer", "const": 1},
        "creative_thesis": {"type": "string", "minLength": 1},
        "visual_strategy": {"type": "string", "minLength": 1},
        "source_strategy": {"type": "string", "minLength": 1},
        "visual_arc": {
            "type": "array", "minItems": 1, "maxItems": 4,
            "items": {"type": "string", "minLength": 1},
        },
        "payoff": {"type": "string", "minLength": 1},
        "avoid": {
            "type": "array", "maxItems": 4,
            "items": {"type": "string", "minLength": 1},
        },
        "fallback": {"type": "string"},
    },
    "required": [
        "contract_version", "creative_thesis", "visual_strategy",
        "source_strategy", "visual_arc", "payoff",
    ],
}

_CANONICAL_IMPLEMENTATION_NAMES = {
    "shorts_fast_pipeline.py",
    "production_guardrails.py",
    "test_shorts_fast_pipeline.py",
}
_CANONICAL_REFERENCE_NAMES = {
    "montaj-fast-path.md",
    "production-profiles.md",
}

_VIDEO_WORDS = (
    "video", "shorts", "short", "reels", "reel", "tiktok", "case-vid"
)
_PRODUCTION_WORDS = (
    "üret", "uret", "oluştur", "olustur", "hazırla", "hazirla", "tamamla", "bitir", "sürdür", "surdur", "devam et", "teslim et",
    "yap", "render", "produce", "create", "revize", "düzenle", "duzenle",
    "düzelt", "duzelt", "fix"
)
_ALT_METHOD_WORDS = (
    "code-driven", "kod tabanlı", "kod tabanli", "programmatic",
    "programatik", "motion graphics", "motion-graphics",
    "sentetik animasyon", "synthetic animation", "remotion"
)
_EXPLICIT_SYSTEM_WORDS = (
    "paket kur", "paket yükle", "paket yukle", "pip install",
    "apt install", "npm install", "bağımlılık kur", "bagimlilik kur",
    "hermes'i güncelle", "hermesi güncelle", "hermes'i değiştir",
    "hermesi değiştir", "plugin kur", "eklenti kur", "vlm kur",
    # English system-maintenance requests must not be mistaken for a normal
    # production job. These phrases authorize only the mutation policy below;
    # all ordinary safety and tool approval rules still apply.
    "hermes self-improvement", "hermes self improvement",
    "video-production system", "video production system",
    "improve hermes", "modify hermes", "change hermes"
)

_INSTALL_RE = re.compile(
    r"(?i)(?:^|[;&|]\s*|\s)"
    r"(?:pip3?\s+install|uv\s+pip\s+install|apt(?:-get)?\s+install|"
    r"npm\s+install|pnpm\s+add|yarn\s+add|conda\s+install)\b"
)


_PROTECTED_MUTATION_RE = re.compile(
    r"(?i)(?:^|[;&|]\s*)(?:sudo\s+)?"
    r"(?:rm|mv|cp|sed\s+-i|tee|chmod|chown|git\s+(?:clone|checkout|reset|pull))\b"
    r"[^;&|]*?(?:/home/hermes/)?\.hermes/(?:hermes-agent|tools|skills|plugins)/"
)

_PROTECTED_PREFIXES = (
    "/home/hermes/.hermes/hermes-agent/",
    "/home/hermes/.hermes/tools/",
    "/home/hermes/.hermes/skills/",
    "/home/hermes/.hermes/plugins/",
)

def _creative_contract_root(session_id: str) -> str:
    sid = str(session_id or "").strip()
    if not sid or not _DB.exists():
        return sid
    try:
        from hermes_state import SessionDB
        with SessionDB(_DB, read_only=True) as db:
            lineage = db.get_compression_lineage(sid)
        return str(lineage[0]) if lineage else sid
    except Exception:
        return sid


def _creative_contract_path(session_id: str) -> Optional[Path]:
    root = _creative_contract_root(session_id)
    return (_CREATIVE_CONTRACT_DIR / f"{root}.json") if root else None


def _save_creative_contract(session_id: str, contract: dict) -> bool:
    path = _creative_contract_path(session_id)
    if path is None:
        return False
    try:
        path.parent.mkdir(parents=True, exist_ok=True)
        payload = {
            "session_root": _creative_contract_root(session_id),
            "saved_at": time.time(),
            "contract": contract,
        }
        tmp = path.with_suffix(".json.tmp")
        tmp.write_text(json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n", encoding="utf-8")
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

root@213-238-170-219:~#