Grill Me

#!/usr/bin/env python3
"""decision_tree_extractor.py — Extract decision branches from a plan/design doc.

Stdlib-only. Scans a markdown plan and identifies decision branches by detecting:

  1. Modal verbs of intent: "we'll", "we will", "we plan to", "we should", "we could"
  2. Open questions: sentences ending in "?"
  3. Choices: "X or Y" / "either X or Y" / "vs"
  4. TBDs: "TBD", "to be decided", "open question"
  5. Trade-off markers: "trade-off", "tradeoff", "pros/cons"

Output: numbered list of decision branches with line refs.

NO LLM CALLS. Pure regex + line walking.

Usage:
    python decision_tree_extractor.py                          # uses embedded sample
    python decision_tree_extractor.py path/to/plan.md
    python decision_tree_extractor.py plan.md --output json
"""

import argparse
import json
import re
import sys
from typing import Any, Dict, List


# Regex patterns that indicate a decision branch
DECISION_PATTERNS = [
    (re.compile(r"\bwe\s*(?:'ll|will|plan\s+to|should|could|might|may)\b", re.IGNORECASE),
     "intent"),
    (re.compile(r"\b(?:either|or)\b.{0,80}\b(?:or|alternatively)\b", re.IGNORECASE),
     "choice"),
    (re.compile(r"\bversus\b|\bvs\.?\b", re.IGNORECASE),
     "choice"),
    (re.compile(r"\bTBD\b|\bto\s+be\s+(?:decided|determined)\b", re.IGNORECASE),
     "open"),
    (re.compile(r"\bopen\s+question\b", re.IGNORECASE),
     "open"),
    (re.compile(r"\btrade-?offs?\b", re.IGNORECASE),
     "tradeoff"),
    (re.compile(r"\bdepends?\s+on\b", re.IGNORECASE),
     "dependency"),
    (re.compile(r"\?\s*$"),
     "question"),
]


SAMPLE_PLAN = """# Plan: Multi-tenant SaaS Migration

## Architecture
We'll move to a single-tenant database per customer. Or maybe we should
do schema-per-tenant for cost. This is a trade-off between isolation and ops cost.

## Auth
TBD: SSO provider — Okta or Auth0?

## Migration sequence
We plan to migrate the largest tenant first. Depends on whether their data fits in 24h.
Open question: rollback strategy?

## Data layer
We could use Postgres logical replication, but we might prefer dual-writes.
Trade-off: complexity vs zero-downtime guarantee.

## Cut-over
Final decision TBD on whether to flip DNS at midnight or use feature flags.
"""


def extract_branches(text: str) -> List[Dict[str, Any]]:
    branches: List[Dict[str, Any]] = []
    seen_lines = set()
    for line_no, line in enumerate(text.splitlines(), start=1):
        for pattern, kind in DECISION_PATTERNS:
            match = pattern.search(line)
            if not match:
                continue
            if line_no in seen_lines:
                continue
            seen_lines.add(line_no)
            branches.append({
                "line": line_no,
                "kind": kind,
                "trigger": match.group(0),
                "context": line.strip()[:160],
            })
            break
    return branches


def analyze(text: str) -> Dict[str, Any]:
    branches = extract_branches(text)
    by_kind: Dict[str, int] = {}
    for b in branches:
        by_kind[b["kind"]] = by_kind.get(b["kind"], 0) + 1
    return {
        "total_branches": len(branches),
        "by_kind": by_kind,
        "branches": branches,
    }


def render_text(r: Dict[str, Any]) -> str:
    lines = []
    lines.append("=" * 72)
    lines.append("DECISION TREE EXTRACTOR")
    lines.append("=" * 72)
    lines.append("")
    lines.append(f"Total decision branches found: {r['total_branches']}")
    lines.append(f"By kind: {r['by_kind']}")
    lines.append("")
    lines.append("-" * 72)
    for i, b in enumerate(r["branches"], start=1):
        lines.append(f"  [{i:2d}] L{b['line']:>4d} ({b['kind']:11s}) {b['context']}")
    return "\n".join(lines)


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Extract decision branches from a plan/design document.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )
    parser.add_argument("path", nargs="?", help="Path to markdown plan (uses embedded sample if omitted)")
    parser.add_argument("--output", choices=("text", "json"), default="text", help="Output format")
    args = parser.parse_args()

    if args.path:
        try:
            with open(args.path, "r", encoding="utf-8") as f:
                text = f.read()
        except (IOError, OSError) as e:
            print(f"error: {e}", file=sys.stderr)
            return 1
    else:
        text = SAMPLE_PLAN

    result = analyze(text)
    if args.output == "json":
        print(json.dumps(result, indent=2))
    else:
        print(render_text(result))
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""grill_session_tracker.py — Track grill-me session state across turns.

Stdlib-only. JSON-backed session storage for the relentless interrogation pattern.
Tracks: questions asked, answers received, recommendations, decisions locked,
remaining branches. Persistence enables resume across sessions.

Storage: ~/.grill_sessions/<session_name>.json

Actions:
  - start <session_name>: initialize new session from plan doc
  - record <session_name> --question-id N --answer "text": record an answer
  - status <session_name>: show progress
  - list: list all sessions
  - close <session_name>: mark complete + summary

NO LLM CALLS. Stdlib only.

Usage:
    python grill_session_tracker.py --action list
    python grill_session_tracker.py --action start --session my-plan --plan path/to/plan.md
    python grill_session_tracker.py --action record --session my-plan --question-id 1 --answer "we chose X"
    python grill_session_tracker.py --action status --session my-plan
    python grill_session_tracker.py --action close --session my-plan
"""

import argparse
import json
import os
import sys
from datetime import datetime
from typing import Any, Dict, List

# Import question generator
_HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, _HERE)
from question_generator import analyze as analyze_plan, SAMPLE_PLAN  # noqa: E402


SESSIONS_DIR = os.path.expanduser("~/.grill_sessions")


def _ensure_dir() -> None:
    os.makedirs(SESSIONS_DIR, exist_ok=True)


def _session_path(name: str) -> str:
    return os.path.join(SESSIONS_DIR, f"{name}.json")


def _load(name: str) -> Dict[str, Any]:
    path = _session_path(name)
    if not os.path.isfile(path):
        return {}
    with open(path, "r", encoding="utf-8") as f:
        return json.load(f)


def _save(name: str, data: Dict[str, Any]) -> None:
    _ensure_dir()
    with open(_session_path(name), "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2)


def start_session(name: str, plan_path: str) -> Dict[str, Any]:
    if plan_path:
        with open(plan_path, "r", encoding="utf-8") as f:
            plan_text = f.read()
    else:
        plan_text = SAMPLE_PLAN
        plan_path = "<embedded sample>"

    plan_analysis = analyze_plan(plan_text)
    session = {
        "name": name,
        "started_at": datetime.now().isoformat(timespec="seconds"),
        "plan_source": plan_path,
        "total_questions": plan_analysis["total_questions"],
        "questions": plan_analysis["questions"],
        "answers": {},  # question_n -> {"answer": str, "recorded_at": iso}
        "status": "active",
    }
    _save(name, session)
    return session


def record_answer(name: str, qid: int, answer: str) -> Dict[str, Any]:
    session = _load(name)
    if not session:
        raise ValueError(f"Session not found: {name}")
    session["answers"][str(qid)] = {
        "answer": answer,
        "recorded_at": datetime.now().isoformat(timespec="seconds"),
    }
    _save(name, session)
    return session


def session_status(name: str) -> Dict[str, Any]:
    session = _load(name)
    if not session:
        return {"error": f"Session not found: {name}"}
    answered = len(session.get("answers", {}))
    total = session.get("total_questions", 0)
    pct = round(100.0 * answered / max(total, 1), 1)
    next_q = None
    for q in session.get("questions", []):
        if str(q["n"]) not in session.get("answers", {}):
            next_q = q
            break
    return {
        "name": session["name"],
        "status": session.get("status", "active"),
        "answered": answered,
        "total": total,
        "percent_complete": pct,
        "next_question": next_q,
        "all_answers": session.get("answers", {}),
    }


def list_sessions() -> List[str]:
    _ensure_dir()
    return sorted(
        os.path.splitext(f)[0]
        for f in os.listdir(SESSIONS_DIR)
        if f.endswith(".json")
    )


def close_session(name: str) -> Dict[str, Any]:
    session = _load(name)
    if not session:
        raise ValueError(f"Session not found: {name}")
    session["status"] = "closed"
    session["closed_at"] = datetime.now().isoformat(timespec="seconds")
    _save(name, session)
    return session


def render_status(r: Dict[str, Any]) -> str:
    if "error" in r:
        return f"ERROR: {r['error']}"
    lines = []
    lines.append("=" * 72)
    lines.append(f"GRILL SESSION: {r['name']}")
    lines.append("=" * 72)
    lines.append(f"Status: {r['status']}  ({r['answered']} / {r['total']} answered, {r['percent_complete']}%)")
    lines.append("")
    if r["next_question"]:
        q = r["next_question"]
        lines.append(f"Next question (Q{q['n']}):")
        lines.append(f"  {q['question']}")
        lines.append(f"  Recommended: {q['recommended']}")
    else:
        lines.append("All questions answered. Run --action close to mark session complete.")
    lines.append("")
    if r["all_answers"]:
        lines.append("Answered:")
        for qid, ans in sorted(r["all_answers"].items(), key=lambda x: int(x[0])):
            lines.append(f"  Q{qid}: {ans['answer'][:100]}")
    return "\n".join(lines)


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Track grill-me session state across turns.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )
    action_choices = ("start", "record", "status", "list", "close")
    parser.add_argument("--action", default="status", choices=action_choices, help="Session action")
    parser.add_argument("--session", help="Session name")
    parser.add_argument("--plan", default="", help="Path to plan markdown (start action)")
    parser.add_argument("--question-id", type=int, help="Question number to record")
    parser.add_argument("--answer", help="Answer text (record action)")
    parser.add_argument("--output", choices=("text", "json"), default="text", help="Output format")
    return parser


def _print_session_list(sessions: List[str], json_output: bool) -> None:
    if json_output:
        print(json.dumps({"sessions": sessions}, indent=2))
        return
    print("Sessions:")
    items = sessions or ["(none)"]
    for s in items:
        print(f"  - {s}")


def _print_start_summary(session: Dict[str, Any]) -> None:
    print(f"Started session: {session['name']}")
    print(f"  Plan: {session['plan_source']}")
    print(f"  Total questions: {session['total_questions']}")
    questions = session.get("questions") or []
    first = questions[0]["question"] if questions else "(none)"
    print(f"  First question: {first}")


def _action_list(args: argparse.Namespace) -> int:
    _print_session_list(list_sessions(), args.output == "json")
    return 0


def _action_start(args: argparse.Namespace) -> int:
    name = args.session or "sample-session"
    session = start_session(name, args.plan)
    if args.output == "json":
        print(json.dumps(session, indent=2))
    else:
        _print_start_summary(session)
    return 0


def _action_record(args: argparse.Namespace) -> int:
    if not args.session or args.question_id is None or not args.answer:
        print("error: record requires --session, --question-id, --answer", file=sys.stderr)
        return 1
    record_answer(args.session, args.question_id, args.answer)
    result = session_status(args.session)
    output = json.dumps(result, indent=2) if args.output == "json" else render_status(result)
    print(output)
    return 0


def _action_status(args: argparse.Namespace) -> int:
    name = args.session or "sample-session"
    result = session_status(name)
    output = json.dumps(result, indent=2) if args.output == "json" else render_status(result)
    print(output)
    return 0


def _action_close(args: argparse.Namespace) -> int:
    if not args.session:
        print("error: close requires --session", file=sys.stderr)
        return 1
    session = close_session(args.session)
    if args.output == "json":
        print(json.dumps(session, indent=2))
    else:
        print(f"Closed session: {args.session}")
    return 0


ACTION_DISPATCH = {
    "list": _action_list,
    "start": _action_start,
    "record": _action_record,
    "status": _action_status,
    "close": _action_close,
}


def main() -> int:
    args = _build_parser().parse_args()
    handler = ACTION_DISPATCH.get(args.action)
    if handler is None:
        return 0
    return handler(args)


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""question_generator.py — Generate forcing questions from extracted decision branches.

Stdlib-only. Takes a plan doc, runs decision_tree_extractor, then generates
forcing questions per Matt Pocock's grill-me discipline:

  - Each question maps to one decision branch
  - Each question proposes a recommended answer
  - Questions ordered by dependency (independent first, dependent last)
  - One question per turn (output is a list, not a paragraph)

Template per question:
  Q: [forcing question]
  Recommended: [recommendation with 1-sentence rationale]

Question templates by branch kind:
  - intent  -> "You said you'll X. Why X and not Y?"
  - choice  -> "Between X and Y, which one and why?"
  - open    -> "X is marked TBD. What's blocking the decision?"
  - tradeoff -> "Trade-off between A and B. Which side are you optimizing for?"
  - dependency -> "X depends on Y. Is Y locked in? If not, ask about Y first."
  - question -> "[original question] — what's your current answer?"

Usage:
    python question_generator.py                          # uses embedded sample
    python question_generator.py path/to/plan.md
    python question_generator.py plan.md --output json
"""

import argparse
import json
import sys
import os
from typing import Any, Dict, List

# Import extractor as a module
_HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, _HERE)
from decision_tree_extractor import extract_branches, SAMPLE_PLAN  # noqa: E402


QUESTION_TEMPLATES = {
    "intent": "Why this approach and not the obvious alternative?",
    "choice": "Which side of the choice, and what's the deciding criterion?",
    "open": "What's blocking this decision? What would unblock it today?",
    "tradeoff": "Which side of the trade-off are you optimizing for, and what's the kill criterion?",
    "dependency": "Is the dependency locked in? If not, that decision comes first.",
    "question": "What's your current best answer, even if uncertain?",
}

RECOMMENDED_TEMPLATES = {
    "intent": "State the alternative explicitly + 1 sentence why you rejected it.",
    "choice": "Pick the option that aligns with the constraint you can't change (budget, deadline, team).",
    "open": "Name the missing input. Estimate when it arrives. Decide now under uncertainty if it won't arrive in time.",
    "tradeoff": "Choose the side that's reversible later. Trade-offs are usually one-way; pick the one with the escape hatch.",
    "dependency": "Resolve the upstream decision first. Then re-evaluate this one.",
    "question": "Even a 60%-confidence answer is better than 'we'll figure it out later'.",
}


def _detect_dependencies(branches: List[Dict[str, Any]]) -> List[int]:
    """Reorder: dependency branches go AFTER what they depend on (best-effort)."""
    dep_indices = [i for i, b in enumerate(branches) if b["kind"] == "dependency"]
    non_dep_indices = [i for i, b in enumerate(branches) if b["kind"] != "dependency"]
    return non_dep_indices + dep_indices


def generate_questions(branches: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    ordered = _detect_dependencies(branches)
    questions: List[Dict[str, Any]] = []
    for n, idx in enumerate(ordered, start=1):
        b = branches[idx]
        q_template = QUESTION_TEMPLATES.get(b["kind"], "What's the current state?")
        r_template = RECOMMENDED_TEMPLATES.get(b["kind"], "State your best answer.")
        questions.append({
            "n": n,
            "line": b["line"],
            "branch_kind": b["kind"],
            "context": b["context"],
            "question": f"L{b['line']}: {b['context']} -> {q_template}",
            "recommended": r_template,
        })
    return questions


def analyze(text: str) -> Dict[str, Any]:
    branches = extract_branches(text)
    questions = generate_questions(branches)
    return {
        "total_questions": len(questions),
        "branch_kinds": sorted(set(b["kind"] for b in branches)),
        "questions": questions,
    }


def render_text(r: Dict[str, Any]) -> str:
    lines = []
    lines.append("=" * 72)
    lines.append("FORCING QUESTION GENERATOR (one at a time, per Matt's grill-me)")
    lines.append("=" * 72)
    lines.append("")
    lines.append(f"Total questions: {r['total_questions']}")
    lines.append(f"Branch kinds: {r['branch_kinds']}")
    lines.append("")
    lines.append("-" * 72)
    for q in r["questions"]:
        lines.append(f"  Q{q['n']:>2d}: {q['question']}")
        lines.append(f"        Recommended: {q['recommended']}")
        lines.append("")
    return "\n".join(lines)


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Generate forcing questions from a plan/design document.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )
    parser.add_argument("path", nargs="?", help="Path to markdown plan (uses embedded sample if omitted)")
    parser.add_argument("--output", choices=("text", "json"), default="text", help="Output format")
    args = parser.parse_args()

    if args.path:
        try:
            with open(args.path, "r", encoding="utf-8") as f:
                text = f.read()
        except (IOError, OSError) as e:
            print(f"error: {e}", file=sys.stderr)
            return 1
    else:
        text = SAMPLE_PLAN

    result = analyze(text)
    if args.output == "json":
        print(json.dumps(result, indent=2))
    else:
        print(render_text(result))
    return 0


if __name__ == "__main__":
    sys.exit(main())