Agent proposals: a plan an AI agent proposes, the user approves in the app, the server runs.
A proposal is <project>/code/ti-toolbox/proposals/<id>.json (ARCHITECTURE ยง6): a title, a
rationale and ordered steps {id, kind, config, subject_ids, after, note, overwrite}. Its
lifecycle is pending -> approved | rejected; while pending the user may edit a step
(edit_step), and what is approved is what runs. proposed_steps keeps the agent's version.
The server queues approved steps itself. On approval every step with no after is queued
at once, with created_by the proposer's and a proposal:<id> tag; a step with after
is queued by this module the moment every job of every step it names has succeeded, and marked
skipped when one of them did not. Dependent steps are deferred rather than queued up front
with job-level after, because submission refuses a job whose inputs are not on disk yet
(tit.jobs.preflight: a flex step after a pre step has no head model at approval time) and
a sim_from_flex step has no electrodes until its flex run exists. sim_from_flex names
either an earlier flex step (config.flex_step) or a finished run (config.flex_run); its
montage and currents come from :func:tit.sim.montage_sources.resolve_flex_simulation, the same
function POST /api/sim-from-flex serves.
Progress: a watcher thread per job manager follows its status stream and advances approved
proposals whenever a job finishes; every record change is published to /ws/jobs subscribers
as {"type": "proposal", "proposal": ...}. Step and proposal states are derived from the
step's jobs at read time, never stored.
view
The wire shape (contract Proposal): the record plus derived step/proposal states.
Source code in tit/server/proposals.py
| def view(record: dict[str, Any], manager: Any) -> dict[str, Any]:
"""The wire shape (contract ``Proposal``): the record plus derived step/proposal states."""
out = json.loads(json.dumps(record))
for step in out["steps"]:
step["state"] = _step_state(record, step, manager)
decision = record["decision"]["state"]
if decision != "approved":
out["status"] = decision
else:
states = [s["state"] for s in out["steps"]]
if any(s in ("waiting", "queued", "running") for s in states):
out["status"] = "running"
elif all(s == "succeeded" for s in states):
out["status"] = "succeeded"
else:
out["status"] = "failed"
out["edited"] = [
(s["config"], s["subject_ids"], s["overwrite"]) for s in record["steps"]
] != [
(s["config"], s["subject_ids"], s["overwrite"])
for s in record["proposed_steps"]
]
return out
|
approve
Approve (after applying edits), then queue every step that waits on nothing.
Every step is re-planned first, so the check runs against the project as it is now; an
error, a missing input of a step that waits on nothing, or an output the step would replace
without overwrite refuses the approval (409) and nothing is queued.
Source code in tit/server/proposals.py
| def approve(
manager: Any,
proposal_id: str,
edits: list[Any] | None = None,
note: str | None = None,
) -> dict[str, Any]:
"""Approve (after applying *edits*), then queue every step that waits on nothing.
Every step is re-planned first, so the check runs against the project as it is now; an
error, a missing input of a step that waits on nothing, or an output the step would replace
without ``overwrite`` refuses the approval (409) and nothing is queued.
"""
with _LOCK:
record = _pending(manager, proposal_id)
for edit in edits or []:
_apply_edit(record, edit, manager)
for step in record["steps"]:
_plan_step(record, step, manager)
problems = _blockers(record)
if problems:
_save(manager.project_dir, record)
_publish(record, manager)
raise HTTPException(status_code=409, detail="; ".join(problems))
record["decision"] = {
"state": "approved",
"at": utcnow_iso(),
"note": _text(note, "note", 2000) or None,
}
_save(manager.project_dir, record)
ensure_watcher(manager)
_advance(record, manager)
return _publish(record, manager)
|
dismiss
Hide a finished proposal (done, rejected or failed) from the default list.
Source code in tit/server/proposals.py
| def dismiss(manager: Any, proposal_id: str) -> dict[str, Any]:
"""Hide a finished proposal (done, rejected or failed) from the default list."""
with _LOCK:
record = _load(manager.project_dir, proposal_id)
status = view(record, manager)["status"]
if status not in ("succeeded", "rejected", "failed"):
raise HTTPException(
status_code=409,
detail=f"proposal {proposal_id} is {status}; only a finished plan can be dismissed",
)
record["dismissed_at"] = utcnow_iso()
_save(manager.project_dir, record)
return _publish(record, manager)
|
run_step
Queue one approved step now -- a retry after it failed, errored or was skipped. Steps
that were skipped because of it wait for it again.
Source code in tit/server/proposals.py
| def run_step(manager: Any, proposal_id: str, step_id: str) -> dict[str, Any]:
"""Queue one approved step now -- a retry after it failed, errored or was skipped. Steps
that were skipped because of it wait for it again."""
with _LOCK:
record = _load(manager.project_dir, proposal_id)
if record["decision"]["state"] != "approved":
raise HTTPException(status_code=409, detail="approve the proposal first")
step = next((s for s in record["steps"] if s["id"] == step_id), None)
if step is None:
raise HTTPException(status_code=404, detail=f"no step {step_id!r}")
state = _step_state(record, step, manager)
if state in ("queued", "running", "succeeded"):
raise HTTPException(status_code=409, detail=f"step {step_id} is {state}")
by_id = {s["id"]: s for s in record["steps"]}
unmet = [
a
for a in step["after"]
if _step_state(record, by_id[a], manager) != "succeeded"
]
if unmet:
raise HTTPException(
status_code=409, detail=f"step {step_id} waits on {', '.join(unmet)}"
)
for later in [step, *_dependants(record, step_id)]:
later.update(job_ids=[], error=None, skipped=None)
record.pop("dismissed_at", None) # running again: it is not finished any more
_submit(record, step, manager)
_save(manager.project_dir, record)
_advance(record, manager)
return _publish(record, manager)
|
advance_all
advance_all(manager: Any, finished_job: str | None = None) -> None
Advance every approved proposal; publish those that changed or own finished_job.
Source code in tit/server/proposals.py
| def advance_all(manager: Any, finished_job: str | None = None) -> None:
"""Advance every approved proposal; publish those that changed or own *finished_job*."""
with _LOCK:
for record in _all(manager.project_dir):
changed = _advance(record, manager)
owns = finished_job is not None and any(
finished_job in s["job_ids"] for s in record["steps"]
)
if changed or owns:
_publish(record, manager)
|
prune
Delete finished proposals (succeeded, rejected or failed) by the job registry's own
retention rule (:func:tit.jobs.registry.expired: older than 30 days, or beyond the newest
200), so a plan's record lasts as long as its jobs'. Pending and running plans are kept.
Returns the removed ids; retention passes keep_count/keep_days/now through.
Source code in tit/server/proposals.py
| def prune(manager: Any, **retention: Any) -> list[str]:
"""Delete finished proposals (succeeded, rejected or failed) by the job registry's own
retention rule (:func:`tit.jobs.registry.expired`: older than 30 days, or beyond the newest
200), so a plan's record lasts as long as its jobs'. Pending and running plans are kept.
Returns the removed ids; *retention* passes ``keep_count``/``keep_days``/``now`` through.
"""
from tit.jobs.registry import expired
with _LOCK:
finished = [
(record["id"], record.get("created_at", ""), record.get("updated_at"))
for record in _all(manager.project_dir)
if view(record, manager)["status"] in ("succeeded", "rejected", "failed")
]
removed = expired(finished, **retention)
for proposal_id in removed:
with contextlib.suppress(OSError):
os.unlink(_path(manager.project_dir, proposal_id))
if removed:
logger.info("proposals: pruned %d old plan(s)", len(removed))
return removed
|
ensure_watcher
ensure_watcher(manager: Any) -> None
Start (once per job manager) the thread that advances proposals as jobs finish, after
pruning old finished ones (the job manager prunes its jobs when it starts).
Source code in tit/server/proposals.py
| def ensure_watcher(manager: Any) -> None:
"""Start (once per job manager) the thread that advances proposals as jobs finish, after
pruning old finished ones (the job manager prunes its jobs when it starts)."""
with _LOCK:
if id(manager) in _WATCHED:
return
_WATCHED.add(id(manager))
prune(manager)
q = manager.subscribe_status()
def watch() -> None:
try:
advance_all(manager) # whatever finished while nobody was watching
while getattr(manager, "_started", True):
try:
job = q.get(timeout=1.0)
except queue.Empty:
continue
if job.get("state") in TERMINAL_STATES:
try:
advance_all(manager, job.get("id"))
except Exception: # noqa: BLE001 - never kill the watcher
logger.exception(
"proposals: advancing after %s failed", job.get("id")
)
finally:
manager.unsubscribe_status(q)
with _LOCK:
_WATCHED.discard(id(manager))
threading.Thread(target=watch, name="tit-proposals", daemon=True).start()
|