Human-in-the-Loop Agent: Approval Interrupts, Durable Pauses, Crash-Safe Tools and Escalation

"""Ledgerly's accounts-payable agent, with a person in the loop. You write the functions marked TODO, one step at a
time. A run is saved to runs/<id>.json after every change, so it can stop, wait hours for a person, and carry on."""
import datetime as dt
import json
import os

import ap_tools as T
import harness as H

SYSTEM = ("You are Ledgerly's accounts-payable assistant. Handle the email you are given with the tools. Look up the "
          "vendor and the invoice before paying, and pay only the invoice total. If an action is rejected, blocked or "
          "fails, do not try to work around it. When you are done, reply to the email saying what you did, then "
          "finish with a one-sentence summary.")
RUNS, APPROVALS, JOURNAL = "runs", "approvals", "journal.jsonl"
TODAY = "2026-09-24"
AUTO_PAY_LIMIT = 1000.0
RECENT_DAYS = 30
READ_ONLY = {"find_vendor", "get_invoice"}


def days_between(a, b):
    """Whole days from date string a to date string b ("2026-09-14", "2026-09-24" -> 10)."""
    return (dt.date.fromisoformat(b[:10]) - dt.date.fromisoformat(a[:10])).days


# ---------- Step 1: which calls need a person ----------

def assess(name, args):
    """{"action": "run" | "ask" | "block", "reason": str} for a tool call:
    read-only tools and reply run; delete_vendor is blocked; update_bank_details asks; pay_invoice asks when the
    amount is over AUTO_PAY_LIMIT or the vendor's bank details changed less than RECENT_DAYS before TODAY, and runs
    otherwise (also when the invoice does not exist: the tool will refuse it); any other tool asks."""
    # TODO (Step 1): run / block / ask by tool; for pay_invoice, float(amount) against AUTO_PAY_LIMIT, then
    # days_between(vendor bank_changed, TODAY) against RECENT_DAYS (T.get_invoice, T.vendor_by_id).
    raise NotImplementedError("Step 1: write assess()")


# ---------- Step 2: pause, and survive the pause ----------

def _write_json(path, obj):
    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
    with open(path + ".tmp", "w") as f:
        json.dump(obj, f, indent=1)
        f.flush()
        os.fsync(f.fileno())
    os.replace(path + ".tmp", path)


def save(run):
    """Write the run to runs/<id>.json so that a crash never leaves a half-written file."""
    _write_json(os.path.join(RUNS, run["id"] + ".json"), run)


def load(run_id):
    return json.load(open(os.path.join(RUNS, run_id + ".json")))


def load_approval(approval_id):
    return json.load(open(os.path.join(APPROVALS, approval_id + ".json")))


def save_approval(a):
    _write_json(os.path.join(APPROVALS, a["id"] + ".json"), a)


def new_run(email):
    """A saved run for one email: {"id": "run-<email id>", "status": "running", "messages": [system SYSTEM, user
    <the email as From/Subject/body text>], "pending": [], "waiting_for": None, "answer": None}."""
    run = {"id": f"run-{email['id']}", "status": "running", "pending": [], "waiting_for": None, "answer": None,
           "messages": [{"role": "system", "content": SYSTEM},
                        {"role": "user", "content": f"Email {email['id']}\nFrom: {email['from']}\n"
                                                    f"Subject: {email['subject']}\n\n{email['body']}"}]}
    save(run)
    return run


def request_approval(run, call, args, reason, now):
    """Save approvals/<run id>.<call id>.json = {"id", "run", "call_id", "tool", "args", "reason", "status": "waiting",
    "created": now}; the run becomes "waiting" with waiting_for set to that id."""
    raise NotImplementedError("request_approval() arrives in Step 2")


def finish_call(run, call, text):
    """Append the tool message for the first pending call and remove it from pending."""
    run["messages"].append({"role": "tool", "tool_call_id": call["id"], "content": text})
    run["pending"].pop(0)


def advance(run, now, max_turns=10):
    """Work until the run is "done" or "waiting", saving after every change.
    With a pending call: parse its arguments, assess() it; "block" -> finish it with "Blocked by policy: <reason>";
    "ask" -> request_approval(), save, return; "run" -> finish it with execute_once(). With nothing pending: after
    max_turns model replies, finish with answer "Stopped after <max_turns> turns."; otherwise ask H.chat(messages,
    T.SPECS): tool calls are appended and become pending; no tool calls means done, with its content as answer."""
    raise NotImplementedError("advance() arrives in Step 2")


# ---------- Step 3: a person decides, the run carries on ----------

def decide(approval_id, decision, by, args=None, comment=""):
    """Record a decision on a waiting approval: "approve", "edit" (approve with changed args) or "reject". Sets
    status to "approved" or "rejected", and decided_by, args (the edited ones for "edit"), comment. Raise ValueError
    if the approval is not waiting or the decision is unknown."""
    raise NotImplementedError("decide() arrives in Step 3")


def resume(run_id, now, max_turns=10):
    """Load a run. If it waits for an approval that is still "waiting", return it unchanged. "approved": run
    execute_once() with the approval's args and finish the call with "Approved by <who>. <result>" (or "Approved by
    <who> with changed arguments <args as JSON>. <result>" when edited). "rejected" or "expired": finish it with
    "Rejected by <who>: <comment>" (who is "policy" for expired). Then set it running and advance()."""
    raise NotImplementedError("resume() arrives in Step 3")


# ---------- Step 4: a crash must never pay twice ----------

def _journal():
    state = {}
    if os.path.exists(JOURNAL):
        for line in open(JOURNAL):
            e = json.loads(line)
            state[e["call_id"]] = e
    return state


def _append_journal(entry):
    with open(JOURNAL, "a") as f:
        f.write(json.dumps(entry) + "\n")
        f.flush()
        os.fsync(f.fileno())


def execute_once(call, args):
    """Run a tool call at most once, whatever crashes. Before running, append {"call_id", "state": "started", "tool",
    "args"} to the journal; after, {"call_id", "state": "done", "result"}. If the journal already has "done" for
    this call id, return its result without running; if only "started", do not run it again: return
    "Error: this call may already have run before a crash; a person must check the ledger"."""
    return T.run_tool(call["function"]["name"], args)  # Step 4 makes this crash-safe


# ---------- Step 5: the morning queue ----------

def waiting_approvals():
    """Every approval still "waiting", oldest first."""
    raise NotImplementedError("waiting_approvals() arrives in Step 5")


def expire(now, max_hours=24):
    """Mark waiting approvals created more than max_hours before now as "expired", with comment "No decision within
    <max_hours> hours; escalated to the finance manager." Return their ids."""
    raise NotImplementedError("expire() arrives in Step 5")


def resume_all(now):
    """resume() every waiting run whose approval is no longer waiting; return {run id: status} for all runs."""
    raise NotImplementedError("resume_all() arrives in Step 5")