Desktop: Phase 4 (batch) broker owns queue + autonomy

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

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")

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 <sop>.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 = ""):

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)

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 <sop>.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}

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):