diff --git a/desktop/broker.js b/desktop/broker.js index c9dfe85..6fbd12a 100644 --- a/desktop/broker.js +++ b/desktop/broker.js @@ -38,16 +38,30 @@ function readBody(req, limit = 1 << 20, timeoutMs = 15000) { // POST action routes -> { argv, input } for the engine CLI. A value that could begin with '-' uses // the --opt=value form so argparse never misparses it as an option; the run id is slug-sanitized // (the engine re-sanitizes). Returns null for an unknown path. +const sanitizeId = (v) => String(v || '').toLowerCase().replace(/[^a-z0-9-]/g, '').replace(/^-+/, '') // slug, no leading '-' (engine re-sanitizes) + function actionRequest(pathname, sopDir, body) { switch (pathname) { case '/api/run': { - const id = String(body.id || '').toLowerCase().replace(/[^a-z0-9-]/g, '').replace(/^-+/, '') - const argv = ['run', sopDir, id] + const argv = ['run', sopDir, sanitizeId(body.id)] if (String(body.mode || '').trim().toLowerCase() === 'prepare') argv.push('--prepare') const inputs = String(body.inputs || '').trim() if (inputs) argv.push('--inputs-stdin') // inputs ride on stdin (unbounded; no argparse misparse) return { argv, input: inputs } } + case '/api/queue': { + const argv = ['queue', sopDir, sanitizeId(body.id), '--scope=' + String(body.scope || 'here')] + // 'Queue here' persists a project: folder that LATER launches the run's session. The broker's + // own cwd (an Electron app dir, maybe /) is NOT a meaningful launch folder, so pass it only when + // the desktop app sets $SMBOS_LAUNCH_CWD; otherwise omit it -> the engine uses None and a + // folder-less SOP just gets no project (never an unrelated directory). + if (process.env.SMBOS_LAUNCH_CWD) argv.push('--launch-cwd=' + process.env.SMBOS_LAUNCH_CWD) + const inputs = String(body.inputs || '').trim() + if (inputs) argv.push('--inputs-stdin') + return { argv, input: inputs } + } + case '/api/autonomy': + return { argv: ['autonomy', sopDir, sanitizeId(body.id), '--level=' + String(body.level || '')] } case '/api/resolve': return { argv: ['resolve', sopDir, '--file=' + String(body.file || ''), '--decision=' + String(body.decision || '')] } case '/api/dequeue': @@ -59,7 +73,7 @@ function actionRequest(pathname, sopDir, body) { } } -const ACTION_PATHS = new Set(['/api/run', '/api/resolve', '/api/dequeue', '/api/task-status']) +const ACTION_PATHS = new Set(['/api/run', '/api/queue', '/api/autonomy', '/api/resolve', '/api/dequeue', '/api/task-status']) const EXIT_STATUS = { 0: 200, 3: 409, 4: 404, 8: 400, 9: 409 } // engine exit code -> HTTP status; anything else -> 500 // GET endpoints the broker answers itself, in FastAPI's response shape (parity-tested against the diff --git a/desktop/broker.test.js b/desktop/broker.test.js index f2165aa..ca81af7 100644 --- a/desktop/broker.test.js +++ b/desktop/broker.test.js @@ -187,6 +187,8 @@ test('POST actions: header-token gated; maps each engine exit code to the HTTP s ' resolve) echo \'{"detail":"nf"}\'; exit 4;;\n' + // -> 404 ' dequeue) echo \'{"status":"dequeued"}\'; exit 0;;\n' + // -> 200 ' task-status) echo \'{"detail":"conflict"}\'; exit 9;;\n' + // -> 409 + ' queue) echo \'{"status":"queued","sop":"x"}\'; exit 0;;\n' + // -> 200 + ' autonomy) echo \'{"id":"x","autonomy":"with_me"}\'; exit 0;;\n' + // -> 200 ' run) case "$3" in\n' + ' refuse) echo \'{"detail":"nope"}\'; exit 3;;\n' + // -> 409 ' boom) echo \'{"detail":"boom"}\'; exit 1;;\n' + // -> 500 @@ -209,7 +211,9 @@ test('POST actions: header-token gated; maps each engine exit code to the HTTP s assert.equal((await post('/api/resolve', '{"file":"x.md","decision":"approve"}', T)).status, 404) // exit 4 -> 404 assert.equal((await post('/api/dequeue', '{"file":"x.md"}', T)).status, 200) // exit 0 -> 200 assert.equal((await post('/api/task-status', '{"task_id":1,"status":"done"}', T)).status, 409) // exit 9 -> 409 - assert.equal((await post('/api/resolve', '{}', {})).status, 401) // every action is token-gated + assert.equal((await post('/api/queue', '{"id":"x"}', T)).status, 200) // dispatched -> 200 + assert.equal((await post('/api/autonomy', '{"id":"x","level":"with_me"}', T)).status, 200) // dispatched -> 200 + assert.equal((await post('/api/autonomy', '{}', {})).status, 401) // every action is token-gated broker.close() } finally { // restore, but DELETE if originally unset (env[x] = undefined would set the string "undefined") diff --git a/scripts/dashboard_app.py b/scripts/dashboard_app.py index ccff4f1..779a58e 100644 --- a/scripts/dashboard_app.py +++ b/scripts/dashboard_app.py @@ -37,6 +37,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parent)) # allow `python3 scripts/dashboard_app.py` import generate_dashboard as gd # collect_pending/collect_queued/parse_candidates (parked-result reads) import run_gate # the run gate + spawn, shared with the engine-action CLI the broker invokes +import sop_writes # set_autonomy (gate + fingerprint re-stamp), shared with the engine-action CLI import smbos_lib as lib import state_store as ss import serve_dashboard as legacy # reuse the daemon's osascript launch + apply_item (launch-coupled) @@ -292,42 +293,6 @@ def _procedures(sop_dir): return sorted(out, key=lambda x: x["title"].lower()) -class _SopDrifted(Exception): - """The SOP body drifted from its recorded stamp; an autonomy write must not bless it.""" - - -def _write_autonomy(sop_path, level): - """Write the `autonomy:` frontmatter field and (re-)STAMP the content_hash, so the owner's - deliberate choice is fingerprint-protected even on a previously-unstamped SOP: a later - out-of-band edit (the body, or a silent flip of the level itself, e.g. with_me -> on_its_own) - then trips the drift gate and the unattended runner refuses it. Setting autonomy via the - authenticated dashboard IS the owner vouching for this content at this level. - - The drift check and the write share ONE read (no TOCTOU window): a STAMPED-but-drifted SOP - raises _SopDrifted (the caller returns 409) rather than letting the re-stamp bless the changed - body; an unstamped SOP has no recorded hash to drift from, so it's stamped fresh. Atomic - replace; the temp file is cleaned up on any failure.""" - text = sop_path.read_text(encoding="utf-8") - meta, body = lib.split_frontmatter(text) - if lib.is_drifted(meta, body): # stamped + body changed out-of-band: refuse, don't re-stamp it - raise _SopDrifted() - new_hash = lib.content_fingerprint(body, {**meta, "autonomy": level}) - # Unique temp name (not a fixed .md.tmp) so two concurrent writes to the SAME SOP can't - # collide on the temp file. Same directory, so os.replace is an atomic rename. - fd, tmp_name = tempfile.mkstemp(prefix=sop_path.name + ".", suffix=".tmp", dir=str(sop_path.parent)) - os.close(fd) - tmp = Path(tmp_name) - try: - tmp.write_text(lib.set_frontmatter_fields(text, {"autonomy": level, "content_hash": new_hash}), - encoding="utf-8") - os.replace(tmp, sop_path) - finally: - try: - tmp.unlink() # gone after a successful os.replace; cleans up an orphan on failure - except OSError: - pass - - async def _body_obj(request): """Parse a POST body as a JSON object or raise 400. Shared by the action endpoints.""" try: @@ -636,25 +601,16 @@ async def set_autonomy(request: Request): SOP is refused (review it first) so the stamp can't bless an out-of-band body edit.""" check(request.headers.get("x-smbos-token", "")) body = await _body_obj(request) - level = str(body.get("level") or "").strip().lower() - if level not in lib.AUTONOMY_LEVELS: - raise HTTPException(status_code=400, detail="unknown autonomy level") - sid = re.sub(r"[^a-z0-9-]", "", str(body.get("id") or "").lower()) - sop = lib.find_sop(sop_dir, sid) if sid else None - if sop is None: - raise HTTPException(status_code=404, detail="unknown procedure") - status = (lib.frontmatter_field(sop, "status") or "").strip().lower() - if level == "on_its_own" and status not in ("active", "trusted"): - raise HTTPException(status_code=409, detail="A draft can't run on its own yet. Verify it " - "with a supervised run first, then it can earn more autonomy.") try: - await asyncio.to_thread(_write_autonomy, sop, level) - except _SopDrifted: # stamped + body changed out-of-band: don't let the stamp bless it - raise HTTPException(status_code=409, detail="This procedure was changed outside the normal " - "save flow. Review it first, then set its autonomy.") + return await asyncio.to_thread(sop_writes.set_autonomy, sop_dir, body.get("id"), body.get("level")) + except sop_writes.BadLevel as exc: + raise HTTPException(status_code=400, detail=str(exc)) + except sop_writes.UnknownSop as exc: + raise HTTPException(status_code=404, detail=str(exc)) + except (sop_writes.DraftNotAllowed, sop_writes.SopDrifted) as exc: + raise HTTPException(status_code=409, detail=str(exc)) except (OSError, ValueError): raise HTTPException(status_code=500, detail="could not save the autonomy setting") - return {"id": sid, "autonomy": level} @app.get("/api/queue") def api_queue(t: str = ""): diff --git a/scripts/engine_action.py b/scripts/engine_action.py index bd104e1..7013972 100644 --- a/scripts/engine_action.py +++ b/scripts/engine_action.py @@ -19,10 +19,41 @@ from pathlib import Path import run_gate +import sop_writes import smbos_lib as lib import state_store as ss +def _queue(args): + raw = sys.stdin.read() if args.inputs_stdin else (args.inputs or "") + inputs = raw.strip() or None + try: + sid, project = lib.queue_run(args.sop_dir, args.id, inputs=inputs, + scope=args.scope or "here", launch_cwd=args.launch_cwd or None) + except ValueError as exc: # unknown task + print(json.dumps({"detail": str(exc)})) + return 8 + print(json.dumps({"status": "queued", "sop": sid, "project": Path(project).name if project else ""})) + return 0 + + +def _autonomy(args): + try: + result = sop_writes.set_autonomy(args.sop_dir, args.id, args.level) + except sop_writes.BadLevel as exc: + print(json.dumps({"detail": str(exc)})) + return 8 + except sop_writes.UnknownSop as exc: + print(json.dumps({"detail": str(exc)})) + return 4 + except (sop_writes.DraftNotAllowed, sop_writes.SopDrifted) as exc: + print(json.dumps({"detail": str(exc)})) + return 9 + # OSError/ValueError on the write -> the top-level handler -> exit 1 -> 500 (matches FastAPI) + print(json.dumps(result)) + return 0 + + def _run(args): raw = sys.stdin.read() if args.inputs_stdin else (args.inputs or "") inputs = raw.strip() or None @@ -115,6 +146,21 @@ def main(argv=None): ts.add_argument("--status", default="") ts.set_defaults(func=_task_status) + q = sub.add_parser("queue", help="enqueue a run for later") + q.add_argument("sop_dir") + q.add_argument("id") + q.add_argument("--inputs", default=None) + q.add_argument("--inputs-stdin", action="store_true") + q.add_argument("--scope", default="here") + q.add_argument("--launch-cwd", dest="launch_cwd", default=None) + q.set_defaults(func=_queue) + + au = sub.add_parser("autonomy", help="set a procedure's autonomy dial") + au.add_argument("sop_dir") + au.add_argument("id") + au.add_argument("--level", default="") + au.set_defaults(func=_autonomy) + args = ap.parse_args(argv) try: return args.func(args) diff --git a/scripts/sop_writes.py b/scripts/sop_writes.py new file mode 100644 index 0000000..972c7f2 --- /dev/null +++ b/scripts/sop_writes.py @@ -0,0 +1,80 @@ +"""Owner-initiated SOP writes, shared (stdlib-only) by the FastAPI dashboard and the engine-action +CLI the Node broker invokes, so both gate + write through ONE implementation -- the trust-critical +fingerprint re-stamp / drift check is never re-implemented in Node. + +set_autonomy raises a typed exception per refusal so each caller maps it to its own status (HTTP for +FastAPI, an exit code for the engine) while sharing the message + the actual write.""" + +import os +import re +import tempfile +from pathlib import Path + +import smbos_lib as lib + + +class SetAutonomyError(Exception): + """Base for an autonomy-write refusal; the message is owner-facing.""" + + +class BadLevel(SetAutonomyError): + pass # -> 400 + + +class UnknownSop(SetAutonomyError): + pass # -> 404 + + +class DraftNotAllowed(SetAutonomyError): + pass # -> 409 (can't grant a draft full autonomy) + + +class SopDrifted(SetAutonomyError): + pass # -> 409 (stamped body changed out-of-band; the stamp must not bless it) + + +def _write_autonomy(sop_path, level): + """Write the `autonomy:` frontmatter field and (re-)STAMP the content_hash, so the owner's + deliberate choice is fingerprint-protected even on a previously-unstamped SOP: a later out-of-band + edit (the body, or a silent flip of the level itself) then trips the drift gate and the unattended + runner refuses it. The drift check + the write share ONE read (no TOCTOU): a STAMPED-but-drifted + SOP raises SopDrifted rather than letting the re-stamp bless the changed body. Atomic replace.""" + text = sop_path.read_text(encoding="utf-8") + meta, body = lib.split_frontmatter(text) + if lib.is_drifted(meta, body): # stamped + body changed out-of-band: refuse, don't re-stamp it + raise SopDrifted("This procedure was changed outside the normal save flow. Review it first, " + "then set its autonomy.") + new_hash = lib.content_fingerprint(body, {**meta, "autonomy": level}) + # Unique temp name (not a fixed .md.tmp) so two concurrent writes to the SAME SOP can't + # collide on the temp file. Same directory, so os.replace is an atomic rename. + fd, tmp_name = tempfile.mkstemp(prefix=sop_path.name + ".", suffix=".tmp", dir=str(sop_path.parent)) + os.close(fd) + tmp = Path(tmp_name) + try: + tmp.write_text(lib.set_frontmatter_fields(text, {"autonomy": level, "content_hash": new_hash}), + encoding="utf-8") + os.replace(tmp, sop_path) + finally: + try: + tmp.unlink() # gone after a successful os.replace; cleans up an orphan on failure + except OSError: + pass + + +def set_autonomy(sop_dir, sop_id, level): + """Validate + gate + write a procedure's autonomy dial. Returns {id, autonomy}. Raises BadLevel / + UnknownSop / DraftNotAllowed / SopDrifted (the caller maps to a status), or OSError on a write + failure. 'On its own' requires an active/trusted SOP -- you can't grant a draft full autonomy.""" + level = str(level or "").strip().lower() + if level not in lib.AUTONOMY_LEVELS: + raise BadLevel("unknown autonomy level") + sid = re.sub(r"[^a-z0-9-]", "", str(sop_id or "").lower()) + sop = lib.find_sop(sop_dir, sid) if sid else None + if sop is None: + raise UnknownSop("unknown procedure") + status = (lib.frontmatter_field(sop, "status") or "").strip().lower() + if level == "on_its_own" and status not in ("active", "trusted"): + raise DraftNotAllowed("A draft can't run on its own yet. Verify it with a supervised run " + "first, then it can earn more autonomy.") + _write_autonomy(sop, level) # raises SopDrifted, or OSError/ValueError on a write failure + return {"id": sid, "autonomy": level} diff --git a/tests/test_engine_action.py b/tests/test_engine_action.py index 8acbba4..22e81f8 100644 --- a/tests/test_engine_action.py +++ b/tests/test_engine_action.py @@ -86,6 +86,43 @@ def test_engine_run_inputs_from_stdin(tmp_path, capsys, monkeypatch): assert captured["inputs"] == "--leading-dash and spaces" # stripped, passed through, not a flag +def test_engine_autonomy_gate_and_write(tmp_path): + (tmp_path / "ops").mkdir() + (tmp_path / "ops" / "act.md").write_text("---\nid: act\ntitle: A\nstatus: active\n---\n# A\n", encoding="utf-8") + (tmp_path / "ops" / "wip.md").write_text("---\nid: wip\ntitle: W\nstatus: draft\n---\n# W\n", encoding="utf-8") + assert engine_action.main(["autonomy", str(tmp_path), "act", "--level=bogus"]) == 8 # 400 bad level + assert engine_action.main(["autonomy", str(tmp_path), "nope", "--level=with_me"]) == 4 # 404 unknown + assert engine_action.main(["autonomy", str(tmp_path), "wip", "--level=on_its_own"]) == 9 # 409 draft + assert engine_action.main(["autonomy", str(tmp_path), "act", "--level=with_me"]) == 0 + import smbos_lib as lib + assert lib.autonomy_level(str(tmp_path), "act") == "with_me" # persisted to frontmatter + + +def test_engine_autonomy_refuses_drift_and_restamps(tmp_path): + # the trust property THROUGH the engine path (stdlib test job): a clean write re-stamps; a body + # that drifted out-of-band makes the next write refuse (SopDrifted -> exit 9 -> 409), not bless it. + import smbos_lib as lib + (tmp_path / "ops").mkdir() + p = tmp_path / "ops" / "act.md" + p.write_text("---\nid: act\ntitle: A\nstatus: active\n---\n# A\nbody\n", encoding="utf-8") + meta, body = lib.split_frontmatter(p.read_text(encoding="utf-8")) + p.write_text(lib.set_frontmatter_fields(p.read_text(encoding="utf-8"), + {"content_hash": lib.content_fingerprint(body, meta)}), encoding="utf-8") # stamp it + assert lib.has_unrecorded_changes(str(tmp_path), "act") is False + assert engine_action.main(["autonomy", str(tmp_path), "act", "--level=prepare_ask"]) == 0 + assert lib.has_unrecorded_changes(str(tmp_path), "act") is False # re-stamped, not drift + p.write_text(p.read_text(encoding="utf-8") + "\nout-of-band edit\n", encoding="utf-8") + assert engine_action.main(["autonomy", str(tmp_path), "act", "--level=on_its_own"]) == 9 # drift -> 409 + + +def test_engine_queue(tmp_path): + (tmp_path / "ops").mkdir() + (tmp_path / "ops" / "act.md").write_text("---\nid: act\ntitle: A\nstatus: active\n---\n# A\n", encoding="utf-8") + assert engine_action.main(["queue", str(tmp_path), "nope"]) == 8 # unknown task -> 400 + assert engine_action.main(["queue", str(tmp_path), "act"]) == 0 + assert any((tmp_path / "queue").glob("*.md")) # a queue file was written + + def test_engine_run_internal_error_is_caught(tmp_path, capsys, monkeypatch): # an unexpected failure in the engine -> exit 1 (the broker maps this to 500), never an unhandled crash def boom(*a, **k):