feat: orc propose — the approval queue's front door

README.md

@@ -72,12 +72,33 @@ orc research "<topic>" [-w <name>]     corpus-grounded synthesis with citations
 orc trace show <run_id>                full trace JSON
 orc trace list [-w <name>]             recent runs
 orc replay <run_id> [--live]           re-execute a recorded run
-orc approve list [-w <name>]           list pending approval items
+orc propose <executor> --params <json>  stage an action for human approval
+orc approve list [-w <name>] [--json]   list pending approval items
 orc approve accept <id> [--note]       accept a pending recommendation
 orc approve reject <id> [--note]       reject one
 orc mcp serve                          start the MCP stdio server
 ```
 
+### Propose from the CLI
+
+Anything that would touch the outside world goes through the approval queue —
+including from the command line:
+
+```bash
+# one-time: enable the executor for the workspace (deny-by-default)
+cat >> ~/.orc/config.toml <<'TOML'
+[workspace.research.effects]
+allowed = ["fs.write_file"]
+TOML
+
+orc propose fs.write_file \
+  --params '{"path": "summary.md", "content": "..."}' \
+  --summary "publish verified summary" -w research
+# -> approval <id> pending; then:
+orc approve accept <id> -w research
+orc execute <id> -w research          # lands in ~/.orc/workspaces/research/out/
+```
+
 ## Architecture
 
 ```

src/orc/cli.py

@@ -6,6 +6,7 @@
 from orc.cli_commands import execute as execute_cmd
 from orc.cli_commands import ingest as ingest_cmd
 from orc.cli_commands import mcp as mcp_cmd
+from orc.cli_commands import propose as propose_cmd
 from orc.cli_commands import replay as replay_cmd
 from orc.cli_commands import research as research_cmd
 from orc.cli_commands import search as search_cmd
@@ -29,6 +30,7 @@ def main() -> None:
 main.add_command(trace_cmd.trace_group)
 main.add_command(replay_cmd.replay_command)
 main.add_command(approve_cmd.approve_group)
+main.add_command(propose_cmd.propose_command)
 main.add_command(execute_cmd.execute_command)
 main.add_command(worker_cmd.worker_command)
 main.add_command(audit_cmd.audit_group)

src/orc/cli_commands/approve.py

@@ -41,7 +41,8 @@ def approve_group() -> None:
     help="Filter by status (default: pending)",
 )
 @click.option("--limit", type=int, default=20)
-def list_command(workspace: str | None, status: str, limit: int) -> None:
+@click.option("--json", "as_json", is_flag=True, help="Machine-readable JSON output")
+def list_command(workspace: str | None, status: str, limit: int, as_json: bool) -> None:
     """List approvals."""
     try:
         ws = ws_module.resolve(workspace)
@@ -50,6 +51,29 @@ def list_command(workspace: str | None, status: str, limit: int) -> None:
     items = approval_module.list_approvals(
         ws.name, status=None if status == "all" else status, limit=limit
     )
+    if as_json:
+        # Plain echo, never rich: scripts (and baton) parse this.
+        click.echo(
+            json_lib.dumps(
+                [
+                    {
+                        "approval_id": a.approval_id,
+                        "status": a.status,
+                        "approvers_required": a.approvers_required,
+                        "accept_count": a.accept_count,
+                        "reject_count": a.reject_count,
+                        "directive": a.directive,
+                        "skill": a.skill,
+                        "summary": a.summary,
+                        "source_run_id": a.source_run_id,
+                        "created_at": a.created_at,
+                    }
+                    for a in items
+                ],
+                indent=2,
+            )
+        )
+        return
     if not items:
         console.print(f"[dim]No approvals with status={status} in {ws.name}[/dim]")
         return

src/orc/cli_commands/propose.py

@@ -0,0 +1,119 @@
+"""`orc propose <executor>` — the approval queue's front door.
+
+Stages an action from the command line through the same guarded path skills use
+(`Run.propose`): executor existence, params schema, and the workspace allow-list
+are all checked *before* anything is enqueued. The CLI never executes — it only
+proposes; a human decision plus `orc execute`/`orc worker` carries it out.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+import click
+
+from orc import effects
+from orc.errors import WorkspaceNotFoundError
+from orc.paths import config_path
+from orc.runs import open_run
+from orc.storage import workspace as ws_module
+
+
+def _load_params(raw: str) -> dict[str, Any]:
+    # curl-style convention: @ means "read the JSON from this file".
+    if raw.startswith("@"):
+        path = Path(raw[1:])
+        if not path.is_file():
+            raise click.ClickException(f"Params file not found: {path}")
+        raw = path.read_text()
+    try:
+        params = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        raise click.ClickException(f"--params is not valid JSON: {exc}") from exc
+    if not isinstance(params, dict):
+        raise click.ClickException("--params must be a JSON object, not a list or scalar")
+    return params
+
+
+@click.command("propose")
+@click.argument("executor_id")
+@click.option("--params", "params_raw", required=True, help="JSON object, or @/path/to/file.json")
+@click.option("--summary", required=True, help="One-line summary shown to approvers")
+@click.option("--workspace", "-w", default=None, help="Workspace name (env: ORC_DEFAULT_WORKSPACE)")
+@click.option("--approvers", type=int, default=1, show_default=True, help="Accepts required")
+@click.option("--idempotency-key", default=None, help="Dedupe key for execution")
+@click.option("--json", "as_json", is_flag=True, help="Machine-readable JSON output")
+def propose_command(
+    executor_id: str,
+    params_raw: str,
+    summary: str,
+    workspace: str | None,
+    approvers: int,
+    idempotency_key: str | None,
+    as_json: bool,
+) -> None:
+    """Propose an action for human approval.
+
+    Note: `orc replay` of a propose-run is unsupported — the "effects" directive
+    is not in the directive registry, and replaying a proposal is semantically
+    wrong (it would silently stage a duplicate effect).
+    """
+    params = _load_params(params_raw)
+    try:
+        ws = ws_module.resolve(workspace)
+    except WorkspaceNotFoundError as exc:
+        raise click.ClickException(str(exc)) from exc
+
+    with open_run(
+        ws,
+        directive="effects",
+        skill="cli.propose",
+        inputs={"executor": executor_id, "summary": summary, "params": params},
+    ) as run:
+        try:
+            approval_id = run.propose(
+                executor=executor_id,
+                params=params,
+                summary=summary,
+                idempotency_key=idempotency_key,
+                approvers_required=approvers,
+            )
+        except effects.ExecutorNotFoundError as exc:
+            known = ", ".join(sorted(e.id for e in effects.all_executors()))
+            raise click.ClickException(f"{exc}. Known executors: {known}") from exc
+        except effects.ExecutorNotAllowedError as exc:
+            raise click.ClickException(
+                f"{exc}\n"
+                f"Enable it by adding to {config_path()}:\n"
+                f"[workspace.{ws.name}.effects]\n"
+                f'allowed = ["{executor_id}"]'
+            ) from exc
+        except effects.ActionValidationError as exc:
+            schema = json.dumps(effects.get(executor_id).params_schema, indent=2)
+            raise click.ClickException(
+                f"Invalid params for {executor_id}: {exc}\nSchema:\n{schema}"
+            ) from exc
+        run.close(output={"approval_id": approval_id})
+        run_id = run.run_id
+
+    if as_json:
+        click.echo(
+            json.dumps(
+                {
+                    "approval_id": approval_id,
+                    "run_id": run_id,
+                    "workspace": ws.name,
+                    "executor": executor_id,
+                    "status": "pending",
+                },
+                indent=2,
+            )
+        )
+    else:
+        click.echo(f"Proposed {executor_id}: approval {approval_id} pending")
+        click.echo("Next steps:")
+        click.echo(f"  orc approve show {approval_id} -w {ws.name}")
+        click.echo(f"  orc approve accept {approval_id} -w {ws.name}")
+        click.echo(f"  orc execute {approval_id} -w {ws.name}")

tests/unit/test_approval_queue.py

@@ -353,3 +353,28 @@ def test_backward_compat_default_single_approver(orc_home: Path) -> None:
     decided = approval_module.accept(name, aid, decided_by="alice")
     assert decided.status == "approved"
     assert decided.progress == "1/1"
+
+
+def test_approve_list_json_emits_machine_readable_array(orc_home: Path) -> None:
+    """Scripts (and baton) need a parseable pending check, not a rich table."""
+    import json as json_lib
+
+    name = _seed_workspace(orc_home)
+    approval_module.enqueue(
+        name,
+        directive="research",
+        skill="t",
+        source_run_id="01HXYZ123",
+        summary="machine readable",
+        payload={},
+    )
+
+    result = CliRunner().invoke(main, ["approve", "list", "-w", name, "--json"])
+    assert result.exit_code == 0, result.output
+    items = json_lib.loads(result.output)
+    assert isinstance(items, list) and len(items) == 1
+    item = items[0]
+    assert item["status"] == "pending"
+    assert item["summary"] == "machine readable"
+    assert {"approval_id", "approvers_required", "accept_count", "reject_count",
+            "directive", "skill", "source_run_id", "created_at"} <= set(item)