"""Architect MCP tool logic (Phase 1, Workstream A/E) — plain, testable functions. The Architect drafts and iterates on the thesis; it CANNOT make anything canonical — promotion to canonical is a human-only action on a CRM HTTP route (server.py), not exposed here (guardrail #4). Every write goes through interaction_log (guardrail #5). Mirrors crm_tools.py conventions. Tool surface: reads list_thesis_lines, get_thesis, get_node, get_node_history, list_versions, get_canonical_thesis, get_review_feedback, list_segments, get_segment drafts create_thesis_line, upsert_thesis_node, create_thesis_version, submit_version_for_review, upsert_segment NO approve/promote/publish/outbound tool exists. """ import json import os import sqlite3 import sys import uuid from datetime import datetime, timezone sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "ingest")) import config # noqa: E402 def _conn(db=None): c = sqlite3.connect(db or os.environ.get("CRM_DB_PATH") or config.DEFAULT_DB) c.row_factory = sqlite3.Row c.execute("PRAGMA foreign_keys=ON") return c def _now(): return datetime.now(timezone.utc).replace(tzinfo=None).isoformat() + "Z" def _eid(prefix): return f"{prefix}_{uuid.uuid4().hex[:16]}" def _log(c, action, target_id, payload, actor_id="architect", actor_type="agent"): c.execute("""INSERT INTO interaction_log (id, ts, actor_type, actor_id, action, target_type, target_id, payload, source, created_at) VALUES (?,?,?,?,?,?,?,?,?,?)""", (str(uuid.uuid4()), _now(), actor_type, actor_id, action, "thesis", target_id, json.dumps(payload) if payload is not None else None, "architect", _now())) def _line_by_key(c, line_key): return c.execute("SELECT * FROM thesis_lines WHERE line_key=? AND deleted_at IS NULL", (line_key,)).fetchone() # ── reads ───────────────────────────────────────────────────────────────────── def list_thesis_lines(db=None): c = _conn(db) rows = [dict(r) for r in c.execute( "SELECT id, line_key, name, segment_key, is_core, status FROM thesis_lines WHERE deleted_at IS NULL ORDER BY is_core DESC, name")] c.close() return {"lines": rows, "count": len(rows)} def _node_tree(c, line_id): nodes = [dict(r) for r in c.execute( "SELECT * FROM thesis_nodes WHERE line_id=? AND deleted_at IS NULL ORDER BY ord", (line_id,))] by_parent = {} for n in nodes: by_parent.setdefault(n["parent_id"], []).append(n) def build(pid): out = [] for n in by_parent.get(pid, []): out.append({**{k: n[k] for k in ("id", "node_type", "title", "body", "status", "variant_group", "ord")}, "children": build(n["id"])}) return out return build(None) def get_thesis(line_key, db=None): """A thesis line + its node tree.""" c = _conn(db) line = _line_by_key(c, line_key) if not line: c.close() return {"error": "not_found", "line_key": line_key} out = {"line": dict(line), "tree": _node_tree(c, line["id"])} c.close() return out def get_node(node_id, db=None): c = _conn(db) r = c.execute("SELECT * FROM thesis_nodes WHERE id=?", (node_id,)).fetchone() c.close() return dict(r) if r else {"error": "not_found", "node_id": node_id} def get_node_variants(node_id, db=None): """All competing options for a node (its variant group). The number is fluid: a node may have one option or many at any moment.""" c = _conn(db) node = c.execute("SELECT line_id, variant_group, node_type, title FROM thesis_nodes WHERE id=?", (node_id,)).fetchone() if not node: c.close() return {"error": "not_found", "node_id": node_id} group = node["variant_group"] or node_id rows = [dict(r) for r in c.execute( "SELECT id, body, title, status, variant_group, meta FROM thesis_nodes " "WHERE (variant_group=? OR id=?) AND deleted_at IS NULL ORDER BY created_at", (group, node_id))] c.close() return {"node_id": node_id, "variant_group": group, "node_type": node["node_type"], "title": node["title"], "variants": rows} def get_node_history(node_id, db=None): c = _conn(db) rows = [dict(r) for r in c.execute( "SELECT rev_no, body, title, status, change_summary, change_reason, actor_type, actor_id, created_at " "FROM thesis_node_revisions WHERE node_id=? ORDER BY rev_no DESC", (node_id,))] c.close() return {"node_id": node_id, "revisions": rows} def list_versions(line_key, db=None): c = _conn(db) line = _line_by_key(c, line_key) if not line: c.close() return {"error": "not_found", "line_key": line_key} rows = [dict(r) for r in c.execute( "SELECT id, version_no, status, rationale, created_by, created_at, approved_at " "FROM thesis_versions WHERE line_id=? ORDER BY version_no DESC", (line["id"],))] c.close() return {"line_key": line_key, "versions": rows} def get_canonical_thesis(line_key, db=None): """The single canonical version's body_json. FAILS CLOSED if none approved — so Scribe/downstream agents can never generate against an unapproved thesis.""" c = _conn(db) line = _line_by_key(c, line_key) if not line: c.close() return {"status": "no_such_line", "line_key": line_key} r = c.execute("SELECT * FROM thesis_versions WHERE line_id=? AND status='canonical'", (line["id"],)).fetchone() c.close() if not r: return {"status": "no_canonical_thesis", "line_key": line_key} return {"status": "ok", "line_key": line_key, "version_id": r["id"], "version_no": r["version_no"], "approved_at": r["approved_at"], "thesis": json.loads(r["body_json"])} def get_review_feedback(version_id, db=None): """Partners' reviews/feedback on a version — what the Architect iterates on.""" c = _conn(db) rows = [dict(r) for r in c.execute( "SELECT reviewer_user_id, decision, feedback, target_node_id, created_at " "FROM thesis_reviews WHERE version_id=? ORDER BY created_at", (version_id,))] approvals = sum(1 for r in rows if r["decision"] == "approve") c.close() return {"version_id": version_id, "reviews": rows, "approvals": approvals} def list_segments(db=None): c = _conn(db) rows = [dict(r) for r in c.execute( "SELECT segment_key, name, definition, needs_to_hear, avoid, version_no FROM segments WHERE status='active' ORDER BY name")] c.close() return {"segments": rows, "count": len(rows)} def get_segment(segment_key, db=None): c = _conn(db) r = c.execute("SELECT * FROM segments WHERE segment_key=? AND status='active'", (segment_key,)).fetchone() c.close() return dict(r) if r else {"error": "not_found", "segment_key": segment_key} # ── draft writes (logged; never canonical) ──────────────────────────────────── def create_thesis_line(line_key, name, segment_key=None, is_core=False, description=None, db=None): c = _conn(db) lid = _eid("thl") c.execute("""INSERT INTO thesis_lines (id, line_key, name, segment_key, is_core, description, created_at, updated_at) VALUES (?,?,?,?,?,?,?,?)""", (lid, line_key, name, segment_key, 1 if is_core else 0, description, _now(), _now())) _log(c, "thesis.line_created", lid, {"line_key": line_key, "segment_key": segment_key, "is_core": bool(is_core)}) c.commit() c.close() return {"id": lid, "line_key": line_key} def upsert_thesis_node(line_id, node_type, body, title=None, parent_id=None, ord=None, variant_group=None, node_id=None, change_reason=None, change_summary=None, actor_id="architect", claude_session_id=None, meta=None, db=None): """Create or edit a node. On edit, the prior state is written to thesis_node_revisions before the live row changes (full provenance).""" c = _conn(db) if node_id: prev = c.execute("SELECT * FROM thesis_nodes WHERE id=?", (node_id,)).fetchone() if not prev: c.close() return {"error": "not_found", "node_id": node_id} rev_no = (c.execute("SELECT COALESCE(MAX(rev_no),0) FROM thesis_node_revisions WHERE node_id=?", (node_id,)).fetchone()[0]) + 1 c.execute("""INSERT INTO thesis_node_revisions (id, node_id, line_id, rev_no, node_type, title, body, status, ord, variant_group, meta, change_summary, change_reason, actor_type, actor_id, claude_session_id, created_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (str(uuid.uuid4()), node_id, prev["line_id"], rev_no, prev["node_type"], prev["title"], prev["body"], prev["status"], prev["ord"], prev["variant_group"], prev["meta"], change_summary, change_reason, "agent", actor_id, claude_session_id, _now())) c.execute("""UPDATE thesis_nodes SET node_type=?, title=COALESCE(?,title), body=?, ord=COALESCE(?,ord), variant_group=?, meta=COALESCE(?,meta), updated_at=? WHERE id=?""", (node_type, title, body, ord, variant_group, json.dumps(meta) if meta else None, _now(), node_id)) _log(c, "thesis.node_revised", node_id, {"line_id": prev["line_id"], "rev_no": rev_no, "reason": change_reason}) out = {"id": node_id, "rev_no": rev_no} else: nid = _eid("thn") if ord is None: ord = (c.execute("SELECT COALESCE(MAX(ord),0) FROM thesis_nodes WHERE line_id=? AND parent_id IS ?", (line_id, parent_id)).fetchone()[0]) + 1.0 c.execute("""INSERT INTO thesis_nodes (id, line_id, parent_id, node_type, ord, title, body, status, variant_group, meta, created_at, updated_at) VALUES (?,?,?,?,?,?,?, 'draft', ?,?,?,?)""", (nid, line_id, parent_id, node_type, ord, title, body, variant_group, json.dumps(meta) if meta else None, _now(), _now())) _log(c, "thesis.node_created", nid, {"line_id": line_id, "node_type": node_type}) out = {"id": nid, "rev_no": 0} c.commit() c.close() return out def create_thesis_version(line_key, rationale=None, created_by="architect", db=None): """Freeze the current node tree of a line into an immutable draft version (body_json = the Architect->Scribe contract). Stays 'draft' until submitted and human-approved.""" c = _conn(db) line = _line_by_key(c, line_key) if not line: c.close() return {"error": "not_found", "line_key": line_key} tree = _node_tree(c, line["id"]) # typed projection for the Scribe contract flat = [dict(r) for r in c.execute( "SELECT node_type, title, body FROM thesis_nodes WHERE line_id=? AND deleted_at IS NULL ORDER BY ord", (line["id"],))] def of(t): return [{"title": n["title"], "body": n["body"]} for n in flat if n["node_type"] == t] body_json = { "line_key": line_key, "name": line["name"], "segment_key": line["segment_key"], "throughline": of("throughline"), "pillars": of("section"), "claims": of("claim"), "proof_points": of("proof_point"), "objections": of("objection"), "segment_cuts": of("segment_cut"), "tree": tree, "generated_at": _now(), } vno = (c.execute("SELECT COALESCE(MAX(version_no),0) FROM thesis_versions WHERE line_id=?", (line["id"],)).fetchone()[0]) + 1 vid = _eid("thv") c.execute("""INSERT INTO thesis_versions (id, line_id, version_no, body_json, status, rationale, created_by, created_at) VALUES (?,?,?,?, 'draft', ?,?,?)""", (vid, line["id"], vno, json.dumps(body_json), rationale, created_by, _now())) _log(c, "thesis.version_created", vid, {"line_key": line_key, "version_no": vno}) c.commit() c.close() return {"id": vid, "version_no": vno, "status": "draft"} def submit_version_for_review(version_id, db=None): c = _conn(db) r = c.execute("SELECT status FROM thesis_versions WHERE id=?", (version_id,)).fetchone() if not r: c.close() return {"error": "not_found", "version_id": version_id} if r["status"] != "draft": c.close() return {"error": "not_draft", "status": r["status"]} c.execute("UPDATE thesis_versions SET status='in_review' WHERE id=?", (version_id,)) _log(c, "thesis.submitted_for_review", version_id, None) c.commit() c.close() return {"version_id": version_id, "status": "in_review"} def upsert_segment(segment_key, name, definition=None, needs_to_hear=None, avoid=None, db=None): """Create/replace a segment's active definition (retire the prior active row).""" c = _conn(db) prev = c.execute("SELECT version_no FROM segments WHERE segment_key=? AND status='active'", (segment_key,)).fetchone() vno = (prev["version_no"] + 1) if prev else 1 if prev: c.execute("UPDATE segments SET status='retired', updated_at=? WHERE segment_key=? AND status='active'", (_now(), segment_key)) sid = _eid("seg") c.execute("""INSERT INTO segments (id, segment_key, name, definition, needs_to_hear, avoid, version_no, status, created_at, updated_at) VALUES (?,?,?,?,?,?,?, 'active', ?,?)""", (sid, segment_key, name, definition, needs_to_hear, avoid, vno, _now(), _now())) _log(c, "segment.upserted", sid, {"segment_key": segment_key, "version_no": vno}) c.commit() c.close() return {"id": sid, "segment_key": segment_key, "version_no": vno}