write_hook.py

#!/usr/bin/env python3
"""
Memory Engineering V1 — Write Hook
对话中产生新信息时调用，走 Trigger→Classify→Lifecycle→Storage 管线。
用法: echo '{"type":"decision","content":"...","source":"..."}' | python3 write_hook.py
      或 python3 write_hook.py '<JSON字符串>'
输出: JSON — 操作结果
同时写入 Atomic Event（Event Layer 集成）。
"""
import json, os, glob, sys, re, uuid
from datetime import datetime

BASE = os.path.expanduser("~/Desktop/Memory_System/memory")
AGENT = "hermes"
SYSTEM_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, SYSTEM_DIR)

# ─── Step 1: Trigger ───────────────────────────────────────────
DROP_PATTERNS = [
    r"^(哈哈|嗯|好的|知道了|OK|ok|好|行|可以|明白了)[\s!！。.]*$",
    r"^(今天|明天|昨天|这周|下周)",
    r"^[?？]|^什么|^怎么[样么]",
]

PASS_PATTERNS = [
    r"(应该|必须|不要|禁止|只能|不允许|强制)",
    r"(选择|决定|方案|用|不用|替代)",
    r"(学到|教训|经验|发现|原来|以后|下次)",
    r"(CiteFlow|Hermes|Memory|FDE|Agent|海老|玄老|药老|风老|游景峰)",
    r"(偏好|喜欢|风格|习惯|方式)",
    r"(完成|通过|失败|bug|修|改)",
]

def trigger_check(event: dict) -> tuple:
    content = event.get("content", "")
    event_type = event.get("type", "")
    for pat in DROP_PATTERNS:
        if re.search(pat, content):
            return False, f"matched DROP pattern: {pat}"
    if event_type == "state_change":
        return True, "state_change always passes"
    for pat in PASS_PATTERNS:
        if re.search(pat, content):
            return True, f"matched PASS pattern: {pat}"
    if len(content) > 30 and any(v in content for v in ["了", "到", "用", "做", "改", "写"]):
        return True, "structural content with verb"
    return False, "no pattern matched"


# ─── 否定词检测 ──────────────────────────────────
NEGATION_WORDS = ['不', '没', '不是', '不要', '不用', '不能', '别', '非', '无', '未', '否']


def has_negation_before(content: str, keyword: str, window: int = 3) -> bool:
    """检查 keyword 在 content 中出现的位置，往前 window 个字符内是否有否定词。"""
    idx = content.find(keyword)
    if idx == -1:
        return False
    prefix = content[max(0, idx - window):idx]
    for neg in NEGATION_WORDS:
        if neg in prefix:
            return True
    return False


def negation_safe_kw_match(content: str, keywords: list[str]) -> bool:
    """检查 content 是否包含 keywords 中的任一关键词，且该关键词前没有否定词。"""
    for kw in keywords:
        if kw in content and not has_negation_before(content, kw):
            return True
    return False


# ─── Step 2: Classify ──────────────────────────────────────────
def classify(event: dict) -> list:
    content = event.get("content", "")
    event_type = event.get("type", "")
    scores = []

    # Identity
    identity_score = 0
    if negation_safe_kw_match(content, ["偏好", "习惯", "喜欢", "风格", "方式", "学习", "视觉"]):
        identity_score += 1.5
    if negation_safe_kw_match(content, ["我", "游景峰"]) and negation_safe_kw_match(content, ["想要", "希望", "觉得", "认为"]):
        identity_score += 1
    if event_type == "preference":
        identity_score += 2
    if identity_score >= 2:
        scores.append(("identity", identity_score))

    # ProjectState
    ps_score = 0
    # 项目名不需要否定检测（"不是 CiteFlow" 仍然是关于 CiteFlow 的讨论）
    if any(kw in content for kw in ["CiteFlow", "Memory Engineering", "FDE", "简历", "Builder"]):
        ps_score += 1
    if any(kw in content for kw in ["完成", "阶段", "blocker", "next", "下一步", "当前", "状态", "进度"]):
        ps_score += 1
    # 行为词需要否定检测
    if negation_safe_kw_match(content, ["改为", "更新", "推进", "开始", "完成"]):
        ps_score += 1
    if event_type == "state_change":
        ps_score += 2
    if ps_score >= 2:
        scores.append(("project_state", ps_score))

    # Decision（"不用"从关键词列表中移除，由否定检测处理）
    dec_score = 0
    if negation_safe_kw_match(content, ["选择", "决定", "方案", "用", "替代", "改"]):
        dec_score += 1.5
    has_alternatives = (
        any(kw in content for kw in ["A", "B", "方案一", "方案二", "vs"]) or
        len(event.get("alternatives", [])) > 0
    )
    if has_alternatives or event_type == "decision":
        dec_score += 1.5
    if any(kw in content for kw in ["以后", "下次", "未来", "从此", "不再"]):
        dec_score += 1
    if dec_score >= 2.5:
        scores.append(("decision", dec_score))

    # Lesson
    les_score = 0
    if negation_safe_kw_match(content, ["学到", "教训", "经验", "发现", "原来"]):
        les_score += 1.5
    if not any(kw in content for kw in ["CiteFlow", "Doctor", "Probe"]):
        les_score += 1
    if any(kw in content for kw in ["Agent", "LLM", "AI", "系统", "架构"]):
        les_score += 1
    if event_type == "lesson":
        les_score += 1
    if les_score >= 2:
        scores.append(("lesson", les_score))

    # SOP/Rule（否定指令词本身就是 rule 信号，不过滤；正指令词需否定检测）
    rule_score = 0
    # 组1: 否定指令词（本身就是 rule，不需否定检测）
    if any(kw in content for kw in ["不要", "禁止", "不允许"]):
        rule_score += 1.5
    # 组2: 正指令词（需要否定检测——"不是必须的"不应加分）
    if negation_safe_kw_match(content, ["必须", "只能", "强制", "永远"]):
        rule_score += 1.5
    if any(kw in content for kw in ["每次", "所有", "任何", "总是", "一直"]):
        rule_score += 1.5
    if event_type == "rule":
        rule_score += 1
    if rule_score >= 2.5:
        scores.append(("sop_rule", rule_score))
    
    scores.sort(key=lambda x: x[1], reverse=True)
    return scores


# ─── 中文停用词表 ──────────────────────────────────
CHINESE_STOP_WORDS = set([
    '的', '了', '是', '在', '和', '也', '就', '都', '而', '及', '与',
    '着', '或', '一个', '没有', '我们', '你们', '他们', '它们', '自己',
    '这', '那', '这个', '那个', '这些', '那些', '什么', '哪', '怎么',
    '如何', '因为', '所以', '但是', '虽然', '如果', '可以', '需要',
    '已经', '还是', '只是', '不是', '就是', '会', '能', '要', '有',
    '对', '从', '到', '让', '给', '被', '把', '向', '以', '为',
    '很', '更', '最', '非常', '比较', '特别', '好', '大', '小',
    '上', '下', '中', '里', '外', '前', '后', '左', '右',
    '来', '去', '做', '说', '看', '想', '知道', '觉得', '认为',
    '使用', '进行', '通过', '根据', '按照', '关于', '对于',
    # JSON 字段名（tokenize 会切开）
    'id', 'type', 'scope', 'status', 'active', 'private', 'shared',
    'created', 'updated', 'content', 'source', 'null', 'true', 'false',
])


# ─── Step 3: Lifecycle ─────────────────────────────────────────
def determine_lifecycle(primary_type: str, event: dict) -> tuple:
    content = event.get("content", "")
    scope = "private"
    if primary_type in ("lesson", "sop_rule") and is_shareable(event):
        scope = "shared"
    
    target_dir = os.path.join(BASE, "shared" if scope == "shared" else AGENT, primary_type)
    if not os.path.isdir(target_dir):
        return "create", None, scope
    
    existing = sorted(glob.glob(os.path.join(target_dir, "*.json")))
    if not existing:
        return "create", None, scope
    
    best_match = None
    best_score = 0
    
    # project_state: 优先按 project_name 精确匹配
    if primary_type == "project_state" and event.get("project_name"):
        target_project = event["project_name"].lower()
        for f in existing:
            with open(f) as fh:
                mem = json.load(fh)
            if mem.get("status") == "archived":
                continue
            if mem.get("project_name", "").lower() == target_project:
                best_match = (f, mem)
                best_score = 1.0
                break
    
    # 通用：按内容关键词重叠匹配（过滤停用词）
    if not best_match:
        for f in existing:
            with open(f) as fh:
                mem = json.load(fh)
            if mem.get("status") == "archived":
                continue
            mem_text = json.dumps(mem, ensure_ascii=False).lower()
            content_lower = content.lower()
            words = set(re.findall(r'[\u4e00-\u9fff]+|[a-zA-Z]+', content_lower))
            mem_words = set(re.findall(r'[\u4e00-\u9fff]+|[a-zA-Z]+', mem_text))
            # 过滤停用词
            words = words - CHINESE_STOP_WORDS
            mem_words = mem_words - CHINESE_STOP_WORDS
            if not words:
                continue
            overlap = len(words & mem_words)
            score = overlap / max(len(words), 1)
            if score > best_score:
                best_score = score
                best_match = (f, mem)
    
    if best_score > 0.6:
        return "update", best_match[0], scope
    elif best_score > 0.3:
        return "merge", best_match[0], scope
    else:
        return "create", None, scope


def is_shareable(event: dict) -> bool:
    content = event.get("content", "")
    project_kw = ["CiteFlow", "Doctor", "Probe", "Analyst", "页面", "组件", "site"]
    return not any(kw in content for kw in project_kw)


# ─── Step 4: Storage ───────────────────────────────────────────
def get_next_sequence(agent: str, mem_type: str, scope: str) -> int:
    target_dir = os.path.join(BASE, "shared" if scope == "shared" else agent, mem_type)
    if not os.path.isdir(target_dir):
        return 1
    existing = glob.glob(os.path.join(target_dir, "*.json"))
    max_seq = 0
    for f in existing:
        name = os.path.basename(f)
        match = re.search(r'-(\d+)\.json$', name)
        if match:
            max_seq = max(max_seq, int(match.group(1)))
    return max_seq + 1


def build_memory_json(mem_type: str, event: dict, scope: str, seq: int, 
                      agent_used: str = None) -> dict:
    now = datetime.now().isoformat()
    agent_id = agent_used or AGENT
    mem_id = f"{agent_id}-{mem_type}-{seq:03d}"
    content = event.get("content", "")
    source = event.get("source", "conversation")
    
    base = {
        "id": mem_id, "type": mem_type, "scope": scope,
        "status": "active", "created_at": now, "updated_at": now
    }
    
    if mem_type == "sop_rule":
        base.update({"rule": content, "trigger": event.get("trigger", ""),
                     "consequence": event.get("consequence", ""), "source": source})
    elif mem_type == "identity":
        subtype = "preference"
        if any(kw in content for kw in ["名字", "叫", "是", "身份", "角色"]):
            subtype = "core_identity"
        elif any(kw in content for kw in ["目标", "方向", "成为"]):
            subtype = "goal"
        base.update({"subtype": subtype, "content": content,
                     "impact_scope": event.get("impact_scope", "personal")})
    elif mem_type == "project_state":
        base.update({"project_name": event.get("project_name", "Unknown"),
                     "current_phase": event.get("current_phase", ""),
                     "blockers": event.get("blockers", []),
                     "next_actions": event.get("next_actions", []),
                     "last_updated": now})
    elif mem_type == "decision":
        base.update({"decision": content, "alternatives": event.get("alternatives", []),
                     "reason": event.get("reason", ""),
                     "constraints": event.get("constraints", ""),
                     "impact_scope": event.get("impact_scope", "project")})
    elif mem_type == "lesson":
        base.update({"lesson": content, "trigger": event.get("trigger", ""),
                     "source": source, "impact": event.get("impact", "token_waste"),
                     "reinforcement_count": 1, "last_reinforced": now})
    return base


def execute_storage(action: str, mem_type: str, event: dict, 
                    scope: str, target_file: str = None) -> dict:
    now = datetime.now().isoformat()
    if action == "update" and target_file:
        with open(target_file) as fh:
            mem = json.load(fh)
        mem["updated_at"] = now
        mem["status"] = "active"
        if mem_type == "project_state":
            if event.get("current_phase"):
                mem["current_phase"] = event["current_phase"]
            if event.get("blockers"):
                mem["blockers"] = event["blockers"]
            if event.get("next_actions"):
                mem["next_actions"] = event["next_actions"]
            mem["last_updated"] = now
        elif mem_type == "lesson":
            mem["reinforcement_count"] = mem.get("reinforcement_count", 0) + 1
            mem["last_reinforced"] = now
            mem["lesson"] = mem.get("lesson", "") + "\n\n更新: " + event.get("content", "")
        else:
            for key in ["decision", "rule", "content"]:
                if key in mem and event.get("content"):
                    mem[key] = mem[key] + "\n\n更新(" + now[:10] + "): " + event.get("content", "")
                    break
        with open(target_file, 'w') as fh:
            json.dump(mem, fh, ensure_ascii=False, indent=2)
        return {"action": "update", "file": target_file, "id": mem.get("id")}
    elif action == "merge" and target_file:
        with open(target_file) as fh:
            mem = json.load(fh)
        mem["updated_at"] = now
        if mem_type == "lesson":
            mem["reinforcement_count"] = mem.get("reinforcement_count", 0) + 1
            mem["last_reinforced"] = now
            mem["lesson"] = mem.get("lesson", "") + "\n\n补充: " + event.get("content", "")
        with open(target_file, 'w') as fh:
            json.dump(mem, fh, ensure_ascii=False, indent=2)
        return {"action": "merge", "file": target_file, "id": mem.get("id")}
    else:
        seq = get_next_sequence(AGENT, mem_type, scope)
        mem = build_memory_json(mem_type, event, scope, seq)
        target_dir = os.path.join(BASE, "shared" if scope == "shared" else AGENT, mem_type)
        os.makedirs(target_dir, exist_ok=True)
        target_path = os.path.join(target_dir, f"{mem['id']}.json")
        with open(target_path, 'w') as fh:
            json.dump(mem, fh, ensure_ascii=False, indent=2)
        return {"action": "create", "file": target_path, "id": mem["id"]}


# ─── Main Pipeline ─────────────────────────────────────────────
def run_pipeline(event: dict) -> dict:
    passed, trigger_reason = trigger_check(event)
    if not passed:
        return {"hook": "write", "agent": AGENT,
                "timestamp": datetime.now().isoformat(),
                "action": "drop", "reason": trigger_reason}
    classifications = classify(event)
    if not classifications:
        return {"hook": "write", "agent": AGENT,
                "timestamp": datetime.now().isoformat(),
                "action": "drop", "reason": "no type reached threshold"}
    primary_type, primary_score = classifications[0]
    action, target_file, scope = determine_lifecycle(primary_type, event)

    # ── Event Layer: snapshot state_before ──
    state_before = _snapshot_memory_type(primary_type, scope)

    result = execute_storage(action, primary_type, event, scope, target_file)

    # ── Event Layer: snapshot state_after + write atomic event ──
    state_after = _snapshot_memory_type(primary_type, scope)
    _try_write_mutation_event(primary_type, action, state_before, state_after)

    return {"hook": "write", "agent": AGENT,
            "timestamp": datetime.now().isoformat(),
            "action": result["action"], "primary_type": primary_type,
            "primary_score": primary_score, "scope": scope,
            "classifications": [{"type": t, "score": s} for t, s in classifications],
            "result": result}


# ── Event Layer: State Snapshot ───────────────────────────────

def _snapshot_memory_type(mem_type, scope):
    """
    读取某个 memory 类型的当前 active 文件，返回分层索引快照。
    结构与 event_writer.index_core_context 的输出一致。
    """
    target_dir = os.path.join(BASE, "shared" if scope == "shared" else AGENT, mem_type)
    if not os.path.isdir(target_dir):
        return {}

    files = sorted(glob.glob(os.path.join(target_dir, "*.json")))
    state = {}

    if mem_type == "identity":
        id_map = {}
        for f in files:
            with open(f) as fh:
                m = json.load(fh)
            if m.get("status") != "active":
                continue
            content = m.get("content", "")
            import hashlib
            h = hashlib.md5(content.encode()).hexdigest()[:8]
            id_map[m["id"]] = {
                "subtype": m.get("subtype", "unknown"),
                "content_hash": h,
                "impact_scope": m.get("impact_scope", "personal")
            }
        if id_map:
            state["identity"] = id_map

    elif mem_type == "project_state":
        ps_map = {}
        for f in files:
            with open(f) as fh:
                m = json.load(fh)
            if m.get("status") != "active":
                continue
            pn = m.get("project_name", "unknown")
            if pn == "Unknown":
                continue
            ps_map[pn] = {
                "phase": (m.get("current_phase", "") or "")[:80],
                "blocker_count": len(m.get("blockers", [])),
                "next_action_count": len(m.get("next_actions", []))
            }
        if ps_map:
            state["project_state"] = ps_map

    elif mem_type == "decision":
        active = [m for f in files 
                  if (lambda m: m.get("status") == "active")(json.load(open(f)))]
        # 重读（上面 lambda 已消费了文件句柄）
        decisions = []
        for f in files:
            with open(f) as fh:
                m = json.load(fh)
            if m.get("status") == "active":
                decisions.append(m)
        if decisions:
            state["decisions"] = {
                "count": len(decisions),
                "latest_ids": [d["id"] for d in decisions[-3:]]
            }

    elif mem_type == "lesson":
        lessons = []
        for f in files:
            with open(f) as fh:
                m = json.load(fh)
            if m.get("status") == "active":
                lessons.append(m)
        if lessons:
            state["lessons"] = {
                "count": len(lessons),
                "latest_ids": [l["id"] for l in lessons[-2:]]
            }

    # sop_rule 暂不索引（rule 变化频率低，V1 跳过）
    return state


def _try_write_mutation_event(mem_type, storage_action, state_before, state_after):
    """尝试写入 memory_mutation Atomic Event。失败不阻塞主流程。"""
    try:
        from event_writer import write_atomic_event
        
        sid = os.environ.get("MEMORY_SESSION_ID")
        if not sid:
            # 从 session_start 写入的临时文件读取
            sid_file = os.path.join(SYSTEM_DIR, ".current_session_id")
            if os.path.exists(sid_file):
                with open(sid_file) as f:
                    sid = f.read().strip()
        if not sid:
            return  # 没有 session_id 说明不在 event 会话中，跳过

        evt = write_atomic_event(
            action="memory_mutation",
            trigger="write_hook.py",
            state_before=state_before,
            state_after=state_after,
            actor="system",
            session_id=sid,
            events_dir=os.path.join(SYSTEM_DIR, "events")
        )
        if evt:
            print(f"[write_hook] Atomic Event written: {evt['id']} (seq={evt['seq']})",
                  file=sys.stderr)
    except Exception as e:
        print(f"[write_hook] Event write failed (non-fatal): {e}", file=sys.stderr)


if __name__ == "__main__":
    if len(sys.argv) > 1:
        raw = " ".join(sys.argv[1:])
    else:
        raw = sys.stdin.read()
    try:
        event = json.loads(raw)
    except json.JSONDecodeError as e:
        print(json.dumps({"hook": "write", "agent": AGENT,
                          "timestamp": datetime.now().isoformat(),
                          "error": f"Invalid JSON: {e}", "input": raw[:200]},
                         ensure_ascii=False, indent=2))
        sys.exit(1)
    output = run_pipeline(event)
    print(json.dumps(output, ensure_ascii=False, indent=2))