덮어쓰기 앞에 사람 세우기

"""쓸 내용을 확정하고, 멱등성 키와 차이를 만든다.

여기서 만든 pending_content 가 승인 화면에 실리고 실행 대상이 된다.
셋이 어긋나지 않게 하는 것이 목적이다. (감사 A-4)
"""
import difflib
import hashlib

HEADER = "# 실행항목\n\n회의록에서 자동으로 뽑았습니다. 사람이 확인한 뒤 쓰십시오.\n\n"


def render_todo(actions):
    """실행항목 목록을 마크다운 문서로 만든다.

    같은 입력이면 같은 문서가 나와야 한다.
    그래야 멱등성 키가 의미를 갖는다.
    """
    lines = [HEADER]
    if not actions:
        lines.append("- (추출된 항목이 없습니다)\n")
    else:
        for action in actions:
            lines.append(
                "- [ ] **%s** — %s (기한 %s)\n"
                % (action["owner"], action["task"], action["due"])
            )
    return "".join(lines)


def make_action_id(note_texts, target_path, requester_id):
    """입력에서 파생한 멱등성 키.

    회의록 내용, 대상 경로, 요청자가 같으면 같은 키가 나온다.
    실행 시점에 무작위로 만들면 중복 실행을 막지 못한다. (감사 A-5)
    """
    material = "|".join(list(note_texts) + [target_path, requester_id])
    return hashlib.sha256(material.encode("utf-8")).hexdigest()[:16]


def make_diff(old, new, target_path):
    """사람이 읽는 차이. 파일이 없으면 빈 문자열."""
    if not old:
        return ""
    lines = difflib.unified_diff(
        old.splitlines(keepends=True),
        new.splitlines(keepends=True),
        fromfile="%s (현재)" % target_path,
        tofile="%s (바뀔 내용)" % target_path,
        n=2,
    )
    return "".join(lines)

def plan_write(state: NotesState) -> dict:
    """쓸 내용을 확정한다. 아직 쓰지 않는다."""
    content = render_todo(state["actions"])
    target = sec_tools.read_workspace_file(state["target_path"])
    exists = not target.startswith("[없음]")
    old = target if exists else ""

    action_id = make_action_id(
        state["note_texts"], state["target_path"], state["requester_id"]
    )
    risk = "high" if exists else "low"
    return {
        "pending_content": content,
        "diff": make_diff(old, content, state["target_path"]),
        "target_exists": exists,
        "risk_level": risk,
        "executed_action_id": action_id,
        "audit_log": [
            "쓸 내용 확정 (risk=%s, action_id=%s). 아직 쓰지 않음" % (risk, action_id)
        ],
    }

def make_action_id(note_texts, target_path, requester_id):
    material = "|".join(list(note_texts) + [target_path, requester_id])
    return hashlib.sha256(material.encode("utf-8")).hexdigest()[:16]

def route_after_plan(state: NotesState) -> Literal["skip", "approval", "auto_approve"]:
    """중복을 먼저 본다. 할 일이 없으면 사람에게 묻지 않는다."""
    if state["executed_action_id"] in _EXECUTED_ACTION_IDS:
        return "skip"
    return "approval" if state["risk_level"] == "high" else "auto_approve"

def approval(state: NotesState) -> dict:
    """덮어쓰기 전에 멈추고 차이를 보여 준다."""
    decision = interrupt(
        {
            "question": "기존 파일을 덮어쓸까요?",
            "target_path": state["target_path"],
            "pending_content": state["pending_content"],
            "diff": state["diff"],
            "options": ["approve", "reject"],
        }
    )
    return {
        "approved": decision == "approve",
        "audit_log": ["사람 결정: %s" % decision],
    }

def write_file(state: NotesState) -> dict:
    if not state["approved"]:
        return {"result": "쓰지 않음", "audit_log": ["미승인으로 종료"]}

    action_id = state["executed_action_id"]
    if action_id in _EXECUTED_ACTION_IDS:
        return {
            "result": "이미 처리됨",
            "audit_log": ["중복 실행 차단 (action_id=%s)" % action_id],
        }

    _EXECUTED_ACTION_IDS.add(action_id)
    written = sec_write_tools.write_workspace_file(
        state["target_path"], state["pending_content"]
    )

"""회의록 정리 Agent.

    collect -> extract -> plan
        |- 이미 처리 -> skip -> END          (승인을 묻지 않는다)
        |- 새 파일   -> auto_approve -> write
        `- 덮어쓰기  -> interrupt(diff 제시)
                          |- reject  -> END
                          `- approve -> write

트랙 4 의 규칙을 그대로 따른다.

  1) 실행 대상(pending_content)을 plan 에서 확정하고, 승인 화면과 실행이 같은 값을 쓴다
  2) 멱등성 키는 입력에서 파생한다
  3) 되돌릴 수 없는 일(덮어쓰기) 앞에 사람이 선다
  4) 할 일이 없으면 승인을 묻지 않는다. 사람의 시간도 자원이다
"""
import json
import os
from typing import Literal

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command

import sec_tools
import sec_write_tools
from notes.notes_plan import make_action_id, make_diff, render_todo
from notes.notes_state import NotesState

BASE_URL = os.environ.get("LOCAL_LLM_BASE_URL", "http://host.docker.internal:1234/v1")
MODEL = "google/gemma-4-e4b"
MAX_TOKENS = 1200

_EXECUTED_ACTION_IDS: set = set()

ACTION_SCHEMA = {
    "type": "object",
    "properties": {
        "actions": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "owner": {"type": "string"},
                    "task": {"type": "string"},
                    "due": {"type": "string"},
                },
                "required": ["owner", "task", "due"],
            },
        }
    },
    "required": ["actions"],
}


def local_extractor(note_texts):
    """로컬 모델로 실행항목을 뽑는다. 요청은 이 기기 밖으로 나가지 않는다."""
    from langchain_openai import ChatOpenAI

    model = ChatOpenAI(
        base_url=BASE_URL,
        api_key="lm-studio",
        model=MODEL,
        temperature=0,
        max_tokens=MAX_TOKENS,
    ).bind(
        response_format={
            "type": "json_schema",
            "json_schema": {"name": "actions", "strict": True, "schema": ACTION_SCHEMA},
        }
    )
    reply = model.invoke(
        [
            ("system", "회의록에서 실행항목을 뽑아 JSON으로만 답한다."),
            ("user", "\n\n".join(note_texts)),
        ]
    )
    if not reply.content:
        raise ValueError("모델 응답이 비어 있습니다. max_tokens 를 늘려 보세요.")
    return json.loads(reply.content)["actions"]


def collect_notes(state: NotesState) -> dict:
    """회의록을 읽는다. 읽기 도구만 쓴다."""
    names = [
        name
        for name in sec_tools.list_workspace_files(state["meeting_dir"])
        if name.endswith(".md")
    ]
    texts = [
        sec_tools.read_workspace_file("%s/%s" % (state["meeting_dir"], name))
        for name in names
    ]
    return {
        "note_names": names,
        "note_texts": texts,
        "audit_log": ["회의록 %d건 읽음: %s" % (len(names), ", ".join(names))],
    }


def make_extract_node(extractor):
    def extract(state: NotesState) -> dict:
        actions = extractor(state["note_texts"])
        return {
            "actions": actions,
            "audit_log": ["실행항목 %d건 추출" % len(actions)],
        }

    return extract


def plan_write(state: NotesState) -> dict:
    """쓸 내용을 확정한다. 아직 쓰지 않는다."""
    content = render_todo(state["actions"])
    target = sec_tools.read_workspace_file(state["target_path"])
    exists = not target.startswith("[없음]")
    old = target if exists else ""

    action_id = make_action_id(
        state["note_texts"], state["target_path"], state["requester_id"]
    )
    risk = "high" if exists else "low"
    return {
        "pending_content": content,
        "diff": make_diff(old, content, state["target_path"]),
        "target_exists": exists,
        "risk_level": risk,
        "executed_action_id": action_id,
        "audit_log": [
            "쓸 내용 확정 (risk=%s, action_id=%s). 아직 쓰지 않음" % (risk, action_id)
        ],
    }


def auto_approve(state: NotesState) -> dict:
    """새 파일 만들기는 자동 승인하되 그 사실을 남긴다."""
    return {
        "approved": True,
        "audit_log": ["새 파일이므로 자동 승인 (사람 개입 없음)"],
    }


def approval(state: NotesState) -> dict:
    """덮어쓰기 전에 멈추고 차이를 보여 준다."""
    decision = interrupt(
        {
            "question": "기존 파일을 덮어쓸까요?",
            "target_path": state["target_path"],
            "pending_content": state["pending_content"],
            "diff": state["diff"],
            "options": ["approve", "reject"],
        }
    )
    return {
        "approved": decision == "approve",
        "audit_log": ["사람 결정: %s" % decision],
    }


def write_file(state: NotesState) -> dict:
    """승인된 경우에만 쓴다. 대상은 승인 화면에 보인 pending_content 다."""
    if not state["approved"]:
        return {"result": "쓰지 않음", "audit_log": ["미승인으로 종료"]}

    action_id = state["executed_action_id"]
    if action_id in _EXECUTED_ACTION_IDS:
        return {
            "result": "이미 처리됨",
            "audit_log": ["중복 실행 차단 (action_id=%s)" % action_id],
        }

    _EXECUTED_ACTION_IDS.add(action_id)
    written = sec_write_tools.write_workspace_file(
        state["target_path"], state["pending_content"]
    )
    note = "백업 %s" % written["backup"] if written["backup"] else "백업 없음 (새 파일)"
    return {
        "backup_name": written["backup"],
        "result": "작성 완료: %s" % state["target_path"],
        "audit_log": ["%s 에 씀. %s" % (state["target_path"], note)],
    }


def skip_duplicate(state: NotesState) -> dict:
    """같은 입력이 이미 처리된 경우. 승인을 묻지 않고 끝낸다."""
    return {
        "result": "이미 처리됨",
        "audit_log": [
            "같은 입력이 이미 처리되어 건너뜀 (action_id=%s)" % state["executed_action_id"]
        ],
    }


def route_after_plan(state: NotesState) -> Literal["skip", "approval", "auto_approve"]:
    """중복을 먼저 본다. 할 일이 없으면 사람에게 묻지 않는다."""
    if state["executed_action_id"] in _EXECUTED_ACTION_IDS:
        return "skip"
    return "approval" if state["risk_level"] == "high" else "auto_approve"


def route_after_approval(state: NotesState) -> Literal["write", "__end__"]:
    return "write" if state["approved"] else "__end__"


def build_graph(extractor=None):
    builder = StateGraph(NotesState)

    builder.add_node("collect", collect_notes)
    builder.add_node("extract", make_extract_node(extractor or local_extractor))
    builder.add_node("plan", plan_write)
    builder.add_node("skip", skip_duplicate)
    builder.add_node("auto_approve", auto_approve)
    builder.add_node("approval", approval)
    builder.add_node("write", write_file)

    builder.add_edge(START, "collect")
    builder.add_edge("collect", "extract")
    builder.add_edge("extract", "plan")
    builder.add_conditional_edges("plan", route_after_plan)
    builder.add_edge("skip", END)
    builder.add_edge("auto_approve", "write")
    builder.add_conditional_edges("approval", route_after_approval)
    builder.add_edge("write", END)

    return builder.compile(checkpointer=InMemorySaver())