From bd6e8325bd920024f2e149c44211506252f3dd60 Mon Sep 17 00:00:00 2001 From: Foster Lipkey Date: Sun, 4 Oct 2026 08:02:29 -0400 Subject: [PATCH 1/2] Add durable atomic approval storage primitives (#522) --- README.md | 2 + autonomy_approvals.py | 189 +++++++++++++++++++++ docs/plans/always-on-policy.md | 8 + tests/test_issue_522_autonomy_approvals.py | 148 ++++++++++++++++ 4 files changed, 347 insertions(+) create mode 100644 autonomy_approvals.py create mode 100644 tests/test_issue_522_autonomy_approvals.py diff --git a/README.md b/README.md index 12dce36d..312e9a30 100644 --- a/README.md +++ b/README.md @@ -225,3 +225,5 @@ Wee's custom bash tool explicitly overrides the SDK's built-in tool, and Wee too ### Always-On development status The opt-in Always-On feature is under staged development (#519). The first slice provides disabled policy primitives for canonical actions and scoped allow/ask/deny JSON rules, with audited creation/revocation metadata. It does not enable autonomous execution or change existing chat behavior. See [policy design and execution surfaces](docs/plans/always-on-policy.md) for remaining enforcement and approval work. + +The next Always-On slice adds isolated durable approval storage (#522): atomic first decisions, expiry, cancellation, replayable owner-filtered events, and single-use reservations. It remains unwired; shared UI approvals and always-allow rule transactions are not yet available. diff --git a/autonomy_approvals.py b/autonomy_approvals.py new file mode 100644 index 00000000..e5df3d0a --- /dev/null +++ b/autonomy_approvals.py @@ -0,0 +1,189 @@ +"""Durable approval storage primitives, not a public API or execution permit. + +Callers must supply server-authenticated ownership and actor identities. This +module persists only action fingerprints, never arguments, credentials or tool +results. Runtime wiring remains disabled until API authorization and action +policy revalidation are implemented. +""" +from contextlib import contextmanager +import math +import os +from pathlib import Path +import sqlite3 +import time +from uuid import uuid4 + +from autonomy_policy import Action, _text + +_DECISIONS = {"approve_once": "approved", "reject": "rejected", "revise": "revision_requested"} + + +class ApprovalConflict(ValueError): + """An idempotency key was reused for different work.""" + + +class ApprovalStore: + def __init__(self, path, *, clock=time.time): + self.path = Path(path) + self.clock = clock + self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + # Create with private permissions before SQLite can write any state. + fd = os.open(self.path, os.O_CREAT | os.O_WRONLY, 0o600) + os.close(fd) + if self.path.stat().st_mode & 0o077: + raise ValueError("Approval database must be private to the service user") + with self._transaction() as db: + version = db.execute('PRAGMA user_version').fetchone()[0] + if version not in (0, 1): + raise ValueError("Unsupported approval schema") + db.execute('''CREATE TABLE IF NOT EXISTS approvals ( + id TEXT PRIMARY KEY, owner TEXT NOT NULL, responsibility TEXT NOT NULL, + intent_key TEXT NOT NULL, fingerprint TEXT NOT NULL, + created_at REAL NOT NULL, expires_at REAL NOT NULL, + status TEXT NOT NULL, decided_by TEXT, decided_at REAL, + decision TEXT, UNIQUE(owner, intent_key))''') + db.execute('''CREATE TABLE IF NOT EXISTS approval_events ( + sequence INTEGER PRIMARY KEY AUTOINCREMENT, + approval_id TEXT NOT NULL, owner TEXT NOT NULL, + kind TEXT NOT NULL, actor TEXT, occurred_at REAL NOT NULL)''') + db.execute('CREATE INDEX IF NOT EXISTS approvals_owner ON approvals(owner, created_at)') + db.execute('CREATE INDEX IF NOT EXISTS events_owner ON approval_events(owner, sequence)') + db.execute('PRAGMA user_version=1') + + @contextmanager + def _transaction(self): + db = sqlite3.connect(self.path, timeout=10, isolation_level=None) + db.row_factory = sqlite3.Row + try: + db.execute('PRAGMA synchronous=FULL') + db.execute('BEGIN IMMEDIATE') + yield db + db.commit() + except BaseException: + db.rollback() + raise + finally: + db.close() + + def _now(self): + now = self.clock() + if type(now) not in (int, float) or not math.isfinite(now) or now < 0: + raise ValueError("Invalid approval clock") + return now + + @staticmethod + def _event(db, row, kind, now, actor=None): + db.execute('INSERT INTO approval_events(approval_id, owner, kind, actor, occurred_at) VALUES(?,?,?,?,?)', + (row['id'], row['owner'], kind, actor, now)) + + @staticmethod + def _row(db, approval_id, owner): + row = db.execute('SELECT * FROM approvals WHERE id=? AND owner=?', (approval_id, owner)).fetchone() + if row is None: + raise KeyError(approval_id) + return dict(row) + + def _expire(self, db, row, now): + if row['status'] in ('pending', 'approved') and row['expires_at'] <= now: + db.execute("UPDATE approvals SET status='expired' WHERE id=?", (row['id'],)) + row['status'] = 'expired' + self._event(db, row, 'expired', now) + return row + + def create(self, action, *, owner, responsibility, intent_key, ttl_seconds=3600): + if not isinstance(action, Action): + raise ValueError("Expected a trusted canonical Action") + for value in (owner, responsibility, intent_key): + _text(value) + if type(ttl_seconds) is not int or not 1 <= ttl_seconds <= 86400: + raise ValueError("Approval expiry must be 1..86400 seconds") + now = self._now() + with self._transaction() as db: + existing = db.execute('SELECT * FROM approvals WHERE owner=? AND intent_key=?', (owner, intent_key)).fetchone() + if existing: + row = dict(existing) + if row['fingerprint'] != action.fingerprint or row['responsibility'] != responsibility: + raise ApprovalConflict("Intent already belongs to a different action or responsibility") + return self._expire(db, row, now) + approval_id = str(uuid4()) + db.execute('''INSERT INTO approvals + (id, owner, responsibility, intent_key, fingerprint, created_at, expires_at, status) + VALUES(?,?,?,?,?,?,?, 'pending')''', + (approval_id, owner, responsibility, intent_key, action.fingerprint, now, now + ttl_seconds)) + row = self._row(db, approval_id, owner) + self._event(db, row, 'created', now) + return row + + def get(self, approval_id, *, owner): + _text(owner) + with self._transaction() as db: + return self._expire(db, self._row(db, approval_id, owner), self._now()) + + def resolve(self, approval_id, *, owner, actor, decision, fingerprint): + """Return (record, won). Only the first pending decision changes state. + + UI submits the fingerprint it reviewed; an old preview cannot authorize + a different action. Always-allow is deliberately unsupported until the + bounded rule-save transaction and its authorization are implemented. + """ + for value in (owner, actor, fingerprint): + _text(value) + if decision not in _DECISIONS: + raise ValueError("Unsupported approval decision") + now = self._now() + with self._transaction() as db: + row = self._expire(db, self._row(db, approval_id, owner), now) + if row['fingerprint'] != fingerprint: + raise ApprovalConflict("Reviewed action fingerprint does not match") + if row['status'] != 'pending': + return row, False + db.execute('UPDATE approvals SET status=?, decided_by=?, decided_at=?, decision=? WHERE id=?', + (_DECISIONS[decision], actor, now, decision, approval_id)) + row = self._row(db, approval_id, owner) + self._event(db, row, row['status'], now, actor) + return row, True + + def cancel(self, approval_id, *, owner, actor): + _text(owner); _text(actor) + now = self._now() + with self._transaction() as db: + row = self._expire(db, self._row(db, approval_id, owner), now) + if row['status'] not in ('pending', 'approved'): + return row, False + db.execute("UPDATE approvals SET status='cancelled', decided_by=?, decided_at=? WHERE id=?", (actor, now, approval_id)) + row = self._row(db, approval_id, owner) + self._event(db, row, 'cancelled', now, actor) + return row, True + + def claim(self, approval_id, action, *, owner, responsibility): + """Reserve once; caller must also revalidate current policy before use. + + A claim is never automatically replayed, including after a restart. It + is a storage reservation, not a guarantee an external effect occurred. + """ + if not isinstance(action, Action): + raise ValueError("Expected a trusted canonical Action") + _text(owner); _text(responsibility) + now = self._now() + with self._transaction() as db: + row = self._expire(db, self._row(db, approval_id, owner), now) + if row['responsibility'] != responsibility: + raise ApprovalConflict("Approval belongs to a different responsibility") + if row['status'] != 'approved': + return False + if row['fingerprint'] != action.fingerprint: + db.execute("UPDATE approvals SET status='invalidated' WHERE id=?", (approval_id,)) + self._event(db, row, 'invalidated', now) + return False + db.execute("UPDATE approvals SET status='claimed' WHERE id=?", (approval_id,)) + self._event(db, row, 'claimed', now) + return True + + def events(self, *, owner, after=0, limit=100): + """Owner-filtered replay cursor; no raw action arguments in event data.""" + _text(owner) + if type(after) is not int or after < 0 or type(limit) is not int or not 1 <= limit <= 100: + raise ValueError("Invalid event cursor or limit") + with self._transaction() as db: + return [dict(row) for row in db.execute('''SELECT sequence, approval_id, kind, actor, occurred_at + FROM approval_events WHERE owner=? AND sequence>? ORDER BY sequence LIMIT ?''', (owner, after, limit))] diff --git a/docs/plans/always-on-policy.md b/docs/plans/always-on-policy.md index 4e56b333..3a99999b 100644 --- a/docs/plans/always-on-policy.md +++ b/docs/plans/always-on-policy.md @@ -15,3 +15,11 @@ PolicyStore persists versioned JSON atomically with restrictive temporary-file p - Direct file/service/message/release integrations: use typed operations and gate at the actual side-effect boundary. Browser actions are opaque until a trusted operation adapter exists. File prefix matching here is lexical: adapters must securely resolve symlinks on the execution host and avoid time-of-check/time-of-use path escapes. Command classification, identity authorization, budget constraints, external audit events, approval expiration and request fanout are not yet implemented. No universal exactly-once guarantee is possible for external actions; uncertain results require reconciliation before retry. + +## Stage 2 storage slice (#522) + +ApprovalStore is an isolated SQLite backend. Requests bind an owner, responsibility, idempotency intent and immutable action fingerprint. Retries with the same intent cannot create new requests; changing action or responsibility under that intent fails. Decisions (approve once, reject, revise), cancellation, expiration, and single-use execution reservations are serialized with SQLite transactions across connections. Requests and monotonically sequenced owner-filtered audit events survive restarts. Raw action arguments, secrets and tool results are not stored or broadcast. A private database file is required. + +A claimed request cannot be automatically replayed after a restart. Claiming is only a reservation; it does not prove a side effect happened and does not replace current policy authorization. A real executor must revalidate identity, policy, scope and request state at the side-effect boundary, record completion/uncertain outcomes, and reconcile an ambiguous external result before any retry. + +Next bounded slice: authenticated owner/approver mapping, request lists and sanitized immutable review previews, reconnectable events, bounded always-allow rule transactions and process-safe coordination with JSON policies. Always-allow is explicitly rejected by the current storage interface until that transaction exists. A transactional outbox/recovery protocol is needed across SQLite and JSON; do not treat separate DB and JSON writes as atomic. Expiry currently advances when a request is accessed; the API/coordinator must sweep and notify waiting requests. Retention controls and schema migration paths are required before deployment. No public endpoints, live UI delivery or runtime wiring are enabled yet. diff --git a/tests/test_issue_522_autonomy_approvals.py b/tests/test_issue_522_autonomy_approvals.py new file mode 100644 index 00000000..13eb1f67 --- /dev/null +++ b/tests/test_issue_522_autonomy_approvals.py @@ -0,0 +1,148 @@ +from concurrent.futures import ThreadPoolExecutor +import json +import os +import threading +import pytest +from autonomy_policy import Action +from autonomy_approvals import ApprovalStore, ApprovalConflict + + +def action(**fields): + return Action(**(dict(agent='wee-dev', operation='service.restart', host='dev', resource='wee') | fields)) + + +def create(store, **fields): + return store.create(action(), **(dict(owner='owner', responsibility='keep-healthy', intent_key='restart-1') | fields)) + + +def resolve(store, request, **fields): + return store.resolve(request['id'], **(dict(owner='owner', actor='mac-user', decision='approve_once', fingerprint=request['fingerprint']) | fields)) + + +def test_request_and_decision_survive_restart(tmp_path): + path = tmp_path/'approvals.sqlite'; store = ApprovalStore(path) + request = create(store) + assert ApprovalStore(path).get(request['id'], owner='owner') == request + record, won = resolve(store, request) + assert won and record['status'] == 'approved' + assert ApprovalStore(path).get(request['id'], owner='owner')['decided_by'] == 'mac-user' + assert [e['kind'] for e in store.events(owner='owner')] == ['created', 'approved'] + + +def test_first_decision_wins_across_store_connections(tmp_path): + path = tmp_path/'approvals.sqlite'; request = create(ApprovalStore(path)) + stores = [ApprovalStore(path) for _ in range(3)] + barrier = threading.Barrier(3) + def decide(index): + barrier.wait() + return resolve(stores[index], request, actor=['mac', 'ios', 'web'][index], decision=['approve_once', 'reject', 'revise'][index]) + with ThreadPoolExecutor(max_workers=3) as pool: + results = list(pool.map(decide, range(3))) + assert sum(won for _, won in results) == 1 + assert len({record['status'] for record, _ in results}) == 1 + assert len(stores[0].events(owner='owner')) == 2 + + +def test_intent_retries_do_not_create_new_requests(tmp_path): + store = ApprovalStore(tmp_path/'db'); request = create(store) + assert create(store) == request + assert len(store.events(owner='owner')) == 1 + with pytest.raises(ApprovalConflict): + store.create(action(host='prod'), owner='owner', responsibility='keep-healthy', intent_key='restart-1') + with pytest.raises(ApprovalConflict): create(store, responsibility='other') + assert create(store, owner='other')['id'] != request['id'] + + +@pytest.mark.parametrize('method', ['get', 'resolve', 'cancel', 'claim']) +def test_wrong_owner_cannot_read_or_change_request(tmp_path, method): + store = ApprovalStore(tmp_path/'db'); request = create(store) + with pytest.raises(KeyError): + if method == 'get': store.get(request['id'], owner='other') + elif method == 'resolve': resolve(store, request, owner='other') + elif method == 'cancel': store.cancel(request['id'], owner='other', actor='web') + else: store.claim(request['id'], action(), owner='other', responsibility='keep-healthy') + assert store.events(owner='other') == [] + assert store.get(request['id'], owner='owner')['status'] == 'pending' + + +def test_fingerprint_mismatch_does_not_approve(tmp_path): + store = ApprovalStore(tmp_path/'db'); request = create(store) + with pytest.raises(ApprovalConflict): resolve(store, request, fingerprint=action(host='prod').fingerprint) + assert store.get(request['id'], owner='owner')['status'] == 'pending' + + +@pytest.mark.parametrize('approved', [False, True]) +def test_expiry_prevents_decision_and_execution(tmp_path, approved): + now = [100]; store = ApprovalStore(tmp_path/'db', clock=lambda: now[0]) + request = create(store, ttl_seconds=10) + if approved: resolve(store, request) + now[0] = 110 + record, won = resolve(store, request) + assert not won and record['status'] == 'expired' + assert not store.claim(request['id'], action(), owner='owner', responsibility='keep-healthy') + assert [e['kind'] for e in store.events(owner='owner')].count('expired') == 1 + + +def test_claim_is_single_use_across_restart(tmp_path): + path = tmp_path/'db'; store = ApprovalStore(path); request = create(store); resolve(store, request) + with ThreadPoolExecutor(max_workers=4) as pool: + claimed = list(pool.map(lambda _: ApprovalStore(path).claim(request['id'], action(), owner='owner', responsibility='keep-healthy'), range(4))) + assert sum(claimed) == 1 + assert not ApprovalStore(path).claim(request['id'], action(), owner='owner', responsibility='keep-healthy') + + +def test_changed_arguments_invalidate_approved_request(tmp_path): + store = ApprovalStore(tmp_path/'db'); request = create(store); resolve(store, request) + assert not store.claim(request['id'], action(arguments_json='{"version":"new"}'), owner='owner', responsibility='keep-healthy') + assert store.get(request['id'], owner='owner')['status'] == 'invalidated' + assert store.events(owner='owner')[-1]['kind'] == 'invalidated' + + +def test_claim_cannot_switch_responsibility(tmp_path): + store = ApprovalStore(tmp_path/'db'); request = create(store); resolve(store, request) + with pytest.raises(ApprovalConflict): store.claim(request['id'], action(), owner='owner', responsibility='other') + assert store.get(request['id'], owner='owner')['status'] == 'approved' + + +@pytest.mark.parametrize('decision,status', [('reject', 'rejected'), ('revise', 'revision_requested')]) +def test_rejection_and_revision_do_not_execute(tmp_path, decision, status): + store = ApprovalStore(tmp_path/'db'); request = create(store) + record, won = resolve(store, request, decision=decision) + assert won and record['status'] == status + assert not store.claim(request['id'], action(), owner='owner', responsibility='keep-healthy') + + +def test_cancel_prevents_late_decision_and_claim(tmp_path): + store = ApprovalStore(tmp_path/'db'); request = create(store); resolve(store, request) + record, won = store.cancel(request['id'], owner='owner', actor='web') + assert won and record['status'] == 'cancelled' + assert not resolve(store, request)[1] + assert not store.claim(request['id'], action(), owner='owner', responsibility='keep-healthy') + + +def test_event_cursor_and_payload_do_not_include_arguments(tmp_path): + store = ApprovalStore(tmp_path/'db') + request = store.create(action(arguments_json='{"private":"synthetic-test-marker"}'), owner='owner', responsibility='r', intent_key='k') + resolve(store, request) + first = store.events(owner='owner', limit=1)[0] + remaining = store.events(owner='owner', after=first['sequence']) + assert [event['kind'] for event in remaining] == ['approved'] + assert 'synthetic-test-marker' not in json.dumps(request) + json.dumps(remaining) + assert b'synthetic-test-marker' not in store.path.read_bytes() + assert store.path.stat().st_mode & 0o077 == 0 + + +@pytest.mark.parametrize('ttl', [0, -1, True, 86401, 1.5]) +def test_invalid_expiry_is_rejected(tmp_path, ttl): + with pytest.raises(ValueError): create(ApprovalStore(tmp_path/'db'), ttl_seconds=ttl) + + +def test_always_allow_not_exposed_before_rule_transaction(tmp_path): + store = ApprovalStore(tmp_path/'db'); request = create(store) + with pytest.raises(ValueError): resolve(store, request, decision='approve_always') + assert store.get(request['id'], owner='owner')['status'] == 'pending' + + +def test_existing_public_database_rejected(tmp_path): + path = tmp_path/'db'; path.touch(mode=0o644) + with pytest.raises(ValueError): ApprovalStore(path) From 6bbac58eb011189f503fabe9e557ce5a22b6c054 Mon Sep 17 00:00:00 2001 From: Foster Lipkey Date: Sun, 4 Oct 2026 09:16:02 -0400 Subject: [PATCH 2/2] feat: shared authenticated approvals and recoverable JSON grants (#522) --- README.md | 2 +- agent_manager.py | 6 + autonomy_approvals.py | 85 +++++++++++- autonomy_policy.py | 54 +++++++- autonomy_service.py | 165 +++++++++++++++++++++++ docs/plans/always-on-policy.md | 23 ++++ tests/test_issue_520_autonomy_policy.py | 2 +- tests/test_issue_522_autonomy_service.py | 94 +++++++++++++ 8 files changed, 420 insertions(+), 11 deletions(-) create mode 100644 autonomy_service.py create mode 100644 tests/test_issue_522_autonomy_service.py diff --git a/README.md b/README.md index 312e9a30..f0052b3a 100644 --- a/README.md +++ b/README.md @@ -224,6 +224,6 @@ Wee's custom bash tool explicitly overrides the SDK's built-in tool, and Wee too ### Always-On development status -The opt-in Always-On feature is under staged development (#519). The first slice provides disabled policy primitives for canonical actions and scoped allow/ask/deny JSON rules, with audited creation/revocation metadata. It does not enable autonomous execution or change existing chat behavior. See [policy design and execution surfaces](docs/plans/always-on-policy.md) for remaining enforcement and approval work. +The opt-in Always-On feature is under staged development (#519). The current foundation provides disabled policy primitives, authenticated shared approval/rule endpoints, replayable events and recoverable bounded JSON grants. It does not enable autonomous execution or change existing chat behavior. See [policy design and execution surfaces](docs/plans/always-on-policy.md) for remaining enforcement and approval work. The next Always-On slice adds isolated durable approval storage (#522): atomic first decisions, expiry, cancellation, replayable owner-filtered events, and single-use reservations. It remains unwired; shared UI approvals and always-allow rule transactions are not yet available. diff --git a/agent_manager.py b/agent_manager.py index a506934b..d1f710b7 100644 --- a/agent_manager.py +++ b/agent_manager.py @@ -13148,6 +13148,12 @@ async def _agents_file_watcher(): lifespan=_lifespan, ) + from autonomy_service import ApprovalService, create_router + autonomy_service = ApprovalService(os.environ.get( + "WEE_ALWAYS_ON_STATE_DIR", str(Path.home() / ".local/state/wee/autonomy"))) + app.include_router(create_router(autonomy_service, authenticate)) + app.state.autonomy_service = autonomy_service + # Expose managers on app.state for testing app.state.bg_task_mgr = bg_task_mgr diff --git a/autonomy_approvals.py b/autonomy_approvals.py index e5df3d0a..26cca99d 100644 --- a/autonomy_approvals.py +++ b/autonomy_approvals.py @@ -6,6 +6,7 @@ policy revalidation are implemented. """ from contextlib import contextmanager +import json import math import os from pathlib import Path @@ -34,7 +35,7 @@ def __init__(self, path, *, clock=time.time): raise ValueError("Approval database must be private to the service user") with self._transaction() as db: version = db.execute('PRAGMA user_version').fetchone()[0] - if version not in (0, 1): + if version not in (0, 1, 2): raise ValueError("Unsupported approval schema") db.execute('''CREATE TABLE IF NOT EXISTS approvals ( id TEXT PRIMARY KEY, owner TEXT NOT NULL, responsibility TEXT NOT NULL, @@ -48,7 +49,12 @@ def __init__(self, path, *, clock=time.time): kind TEXT NOT NULL, actor TEXT, occurred_at REAL NOT NULL)''') db.execute('CREATE INDEX IF NOT EXISTS approvals_owner ON approvals(owner, created_at)') db.execute('CREATE INDEX IF NOT EXISTS events_owner ON approval_events(owner, sequence)') - db.execute('PRAGMA user_version=1') + columns = {row[1] for row in db.execute('PRAGMA table_info(approvals)')} + if 'preview_json' not in columns: + db.execute("ALTER TABLE approvals ADD COLUMN preview_json TEXT NOT NULL DEFAULT '{}'") + db.execute("ALTER TABLE approvals ADD COLUMN scope_json TEXT NOT NULL DEFAULT '{}'") + db.execute('CREATE TABLE IF NOT EXISTS approval_rule_outbox (approval_id TEXT PRIMARY KEY, rule_id TEXT NOT NULL, actor TEXT NOT NULL, scope_json TEXT NOT NULL, published INTEGER NOT NULL DEFAULT 0)') + db.execute('PRAGMA user_version=2') @contextmanager def _transaction(self): @@ -90,13 +96,19 @@ def _expire(self, db, row, now): self._event(db, row, 'expired', now) return row - def create(self, action, *, owner, responsibility, intent_key, ttl_seconds=3600): + def create(self, action, *, owner, responsibility, intent_key, ttl_seconds=3600, preview=None): if not isinstance(action, Action): raise ValueError("Expected a trusted canonical Action") for value in (owner, responsibility, intent_key): _text(value) if type(ttl_seconds) is not int or not 1 <= ttl_seconds <= 86400: raise ValueError("Approval expiry must be 1..86400 seconds") + # Preview is deliberately supplied by a trusted adapter, never copied from arguments. + preview = preview or {'summary': action.operation} + if not isinstance(preview, dict) or set(preview) != {'summary'}: + raise ValueError('Expected a sanitized summary') + _text(preview['summary']) + scope = {key: getattr(action, key) for key in ('agent', 'operation', 'host', 'resource')} now = self._now() with self._transaction() as db: existing = db.execute('SELECT * FROM approvals WHERE owner=? AND intent_key=?', (owner, intent_key)).fetchone() @@ -110,6 +122,7 @@ def create(self, action, *, owner, responsibility, intent_key, ttl_seconds=3600) (id, owner, responsibility, intent_key, fingerprint, created_at, expires_at, status) VALUES(?,?,?,?,?,?,?, 'pending')''', (approval_id, owner, responsibility, intent_key, action.fingerprint, now, now + ttl_seconds)) + db.execute('UPDATE approvals SET preview_json=?, scope_json=? WHERE id=?', (json.dumps(preview), json.dumps(scope), approval_id)) row = self._row(db, approval_id, owner) self._event(db, row, 'created', now) return row @@ -179,6 +192,72 @@ def claim(self, approval_id, action, *, owner, responsibility): self._event(db, row, 'claimed', now) return True + def list(self, *, owner, limit=100): + _text(owner) + if type(limit) is not int or not 1 <= limit <= 100: + raise ValueError('Invalid list limit') + with self._transaction() as db: + # Expire all eligible rows, including those beyond the visible page. + now = self._now() + for row in db.execute("SELECT * FROM approvals WHERE owner=? AND status IN ('pending','approved') AND expires_at<=?", (owner, now)).fetchall(): + self._expire(db, dict(row), now) + return [dict(row) for row in db.execute('SELECT * FROM approvals WHERE owner=? ORDER BY created_at DESC LIMIT ?', (owner, limit))] + + def resolve_always(self, approval_id, *, owner, actor, fingerprint): + """Commit decision and durable rule intent together; publish before approving. + + JSON publication is recoverable and idempotent, not falsely described as + an atomic transaction with SQLite. A pending publication cannot execute. + """ + from autonomy_policy import _KNOWN, _OPAQUE + for value in (owner, actor, fingerprint): + _text(value) + with self._transaction() as db: + now = self._now() + row = self._expire(db, self._row(db, approval_id, owner), now) + if row['fingerprint'] != fingerprint: + raise ApprovalConflict('Reviewed action fingerprint does not match') + scope = json.loads(row['scope_json']) + if scope.get('operation') not in _KNOWN or scope['operation'] in _OPAQUE: + raise ValueError('Opaque or unknown actions cannot receive permanent grants') + if row['status'] != 'pending': + return row, False + db.execute("UPDATE approvals SET status='rule_pending', decided_by=?, decided_at=?, decision='approve_always' WHERE id=?", (actor, now, approval_id)) + db.execute('INSERT INTO approval_rule_outbox(approval_id, rule_id, actor, scope_json) VALUES(?,?,?,?)', (approval_id, str(uuid4()), actor, row['scope_json'])) + row = self._row(db, approval_id, owner) + self._event(db, row, 'rule_pending', now, actor) + return row, True + + def publish_rules(self, policy): + # All integration operations acquire policy before the SQLite lock. + with policy.locked(), self._transaction() as db: + for intent in db.execute('SELECT * FROM approval_rule_outbox WHERE published=0').fetchall(): + row = self._row(db, intent['approval_id'], db.execute('SELECT owner FROM approvals WHERE id=?', (intent['approval_id'],)).fetchone()[0]) + rule = policy.add(actor=intent['actor'], approval_id=row['id'], decision='allow', rule_id=intent['rule_id'], **json.loads(intent['scope_json'])) + db.execute('UPDATE approval_rule_outbox SET published=1 WHERE approval_id=?', (row['id'],)) + # Revocation during crash recovery must never recreate or reactivate a grant. + status = 'approved' if rule.enabled and row['expires_at'] > self._now() else 'expired' + if row['status'] == 'rule_pending': + db.execute('UPDATE approvals SET status=? WHERE id=?', (status, row['id'])) + self._event(db, row, status, self._now(), intent['actor']) + + def finish(self, approval_id, *, owner, outcome): + if outcome not in ('succeeded', 'failed', 'uncertain'): + raise ValueError('Invalid execution outcome') + with self._transaction() as db: + row = self._row(db, approval_id, owner) + if row['status'] != 'claimed': + raise ApprovalConflict('Only claimed work can finish') + db.execute('UPDATE approvals SET status=? WHERE id=?', (outcome, approval_id)) + self._event(db, row, outcome, self._now()) + + def recover_claims(self): + """On exclusive worker startup, abandon uncertain reservations; never replay.""" + with self._transaction() as db: + for row in db.execute("SELECT * FROM approvals WHERE status='claimed'").fetchall(): + db.execute("UPDATE approvals SET status='uncertain' WHERE id=?", (row['id'],)) + self._event(db, dict(row), 'uncertain', self._now()) + def events(self, *, owner, after=0, limit=100): """Owner-filtered replay cursor; no raw action arguments in event data.""" _text(owner) diff --git a/autonomy_policy.py b/autonomy_policy.py index 65b504ae..bb0155f5 100644 --- a/autonomy_policy.py +++ b/autonomy_policy.py @@ -6,6 +6,8 @@ """ from dataclasses import dataclass, asdict from datetime import datetime, timezone +from contextlib import contextmanager +import fcntl import hashlib import json import os @@ -128,7 +130,7 @@ def evaluate(action, rules, *, enabled=False): class PolicyStore: - """Single API-process writer. Multi-process locking is required before use. + """Process-safe JSON policy writer with private advisory locks. Read errors propagate so malformed policies cannot silently become grants. Action arguments/credentials are never persisted in the rule file. @@ -136,9 +138,43 @@ class PolicyStore: def __init__(self, path): self.path = Path(path) self._lock = threading.RLock() + self._depth = threading.local() - def load(self): + @contextmanager + def locked(self): + """Serialize read/modify/write across service processes, including nested calls.""" with self._lock: + depth = getattr(self._depth, "value", 0) + if depth: + self._depth.value += 1 + try: + yield + finally: + self._depth.value -= 1 + return + self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + fd = os.open(str(self.path) + ".lock", os.O_CREAT | os.O_RDWR, 0o600) + try: + fcntl.flock(fd, fcntl.LOCK_EX) + self._depth.value = 1 + yield + finally: + self._depth.value = 0 + fcntl.flock(fd, fcntl.LOCK_UN) + os.close(fd) + + def set_enabled(self, enabled): + if type(enabled) is not bool: + raise ValueError("Expected boolean feature flag") + with self.locked(): + data = self.load() + data['enabled'] = enabled + data['revision'] += 1 + self._save(data) + return data + + def load(self): + with self.locked(): if not self.path.exists(): return {"version": 1, "revision": 0, "enabled": False, "rules": []} data = json.loads(self.path.read_text()) @@ -166,19 +202,25 @@ def _save(self, data): if temporary and os.path.exists(temporary): os.unlink(temporary) - def add(self, *, actor, approval_id, agent, operation, host, resource, decision, path_prefix=False): - with self._lock: + def add(self, *, actor, approval_id, agent, operation, host, resource, decision, path_prefix=False, rule_id=None): + with self.locked(): data = self.load() + if rule_id is not None: + for existing in data['rules']: + if existing.id == rule_id: + if (existing.agent, existing.operation, existing.host, existing.resource, existing.decision, existing.source_approval_id, existing.path_prefix) != (agent, operation, host, resource, decision, approval_id, path_prefix): + raise ValueError('Rule ID belongs to another grant') + return existing if len(data['rules']) >= 1000: raise ValueError("Policy rule limit reached") - rule = Rule(str(uuid4()), agent, operation, host, resource, decision, + rule = Rule(rule_id or str(uuid4()), agent, operation, host, resource, decision, _text(actor), datetime.now(timezone.utc).isoformat(), _text(approval_id), path_prefix=path_prefix) data['rules'].append(rule); data['revision'] += 1 self._save(data) return rule def revoke(self, rule_id, *, actor): - with self._lock: + with self.locked(): data = self.load() for index, rule in enumerate(data['rules']): if rule.id == rule_id: diff --git a/autonomy_service.py b/autonomy_service.py new file mode 100644 index 00000000..2698769d --- /dev/null +++ b/autonomy_service.py @@ -0,0 +1,165 @@ +"""Authenticated shared approvals and the sole boundary for Always-On adapters. + +Ordinary chat runtimes are separate. Always-On workers may only call registered +structured adapters here; shell, browser, SDK built-ins and delegation are not +registered and cannot escape this boundary. +""" +from dataclasses import asdict +import json +from pathlib import Path +from autonomy_approvals import ApprovalStore, ApprovalConflict +from autonomy_policy import Action, PolicyStore, evaluate, _KNOWN, _OPAQUE, _text + +OWNER = 'api-account' + + +def principal(auth): + # Header-supplied identity/channel must never grant authority or forge audit identity. + if auth.get('auth_type') == 'shared_key': + return OWNER, 'shared-key-client' + if auth.get('auth_type') == 'session_token': + return OWNER, 'paired:' + _text(auth['identity']) + raise PermissionError('Authenticated API account required') + + +def public_record(row): + return {**{key: row[key] for key in ('id', 'responsibility', 'fingerprint', 'created_at', 'expires_at', 'status', 'decided_by', 'decided_at', 'decision')}, + 'preview': json.loads(row['preview_json']), 'scope': json.loads(row['scope_json'])} + + +class ApprovalService: + def __init__(self, directory): + self.policy = PolicyStore(Path(directory) / 'action-rules.json') + self.approvals = ApprovalStore(Path(directory) / 'approvals.sqlite') + + def recover_rules(self): + self.approvals.publish_rules(self.policy) + + def decide(self, approval_id, auth, decision, fingerprint): + owner, actor = principal(auth) + with self.policy.locked(): + row = self.approvals.get(approval_id, owner=owner) + scope = json.loads(row['scope_json']) + if not scope: + raise ValueError('Legacy request has no reviewable scope') + data = self.policy.load() + # Disabled mode and deny rules cannot be overridden by any approval. + verdict, reason = evaluate(Action(**scope), data['rules'], enabled=data['enabled']) + if verdict == 'deny' and decision in ('approve_once', 'approve_always'): + raise ApprovalConflict('Current policy blocks execution: ' + reason) + if decision == 'approve_always': + row, won = self.approvals.resolve_always(approval_id, owner=owner, actor=actor, fingerprint=fingerprint) + self.recover_rules() + row = self.approvals.get(approval_id, owner=owner) + else: + row, won = self.approvals.resolve(approval_id, owner=owner, actor=actor, decision=decision, fingerprint=fingerprint) + return {'request': public_record(row), 'won': won} + + def execute(self, action, *, responsibility, intent_key, adapter, approval_id=None, summary): + """Run a trusted synchronous adapter under current policy and single-use claim. + + Adapter must bind the canonical action to the actual operation; no model + labels or generic CLI invocation may supply this callable. Keep external + calls bounded. Lock prevents concurrent local policy revocation mid-start. + """ + if action.operation not in _KNOWN or action.operation in _OPAQUE: + raise PermissionError('No Always-On adapter for this capability') + with self.policy.locked(): + data = self.policy.load() + verdict, reason = evaluate(action, data['rules'], enabled=data['enabled']) + if verdict == 'deny': + return {'status': 'denied', 'reason': reason} + # Every execution reserves a durable intent, including policy-allowed work. + row = self.approvals.create(action, owner=OWNER, responsibility=responsibility, intent_key=intent_key, preview={'summary': summary}) + if approval_id and row['id'] != approval_id: + raise ApprovalConflict('Approval does not belong to this intent') + if verdict == 'allow' and row['status'] == 'pending': + self.approvals.resolve(row['id'], owner=OWNER, actor='policy', decision='approve_once', fingerprint=action.fingerprint) + if not self.approvals.claim(row['id'], action, owner=OWNER, responsibility=responsibility): + return {'status': self.approvals.get(row['id'], owner=OWNER)['status'], 'approval_id': row['id']} + try: + result = adapter(action) + except Exception: + # A failure may have happened after an external side effect. + self.approvals.finish(row['id'], owner=OWNER, outcome='uncertain') + raise + self.approvals.finish(row['id'], owner=OWNER, outcome='succeeded') + return {'status': 'succeeded', 'approval_id': row['id'], 'result': result} + + +def create_router(service, authenticate): + from fastapi import APIRouter, Depends, HTTPException, Query + from pydantic import BaseModel, ConfigDict + router = APIRouter(prefix='/api/v1/autonomy', tags=['Always-On']) + + class Decision(BaseModel): + model_config = ConfigDict(extra='forbid') + decision: str + fingerprint: str + + class RuleInput(BaseModel): + model_config = ConfigDict(extra='forbid') + agent: str + operation: str + host: str + resource: str + decision: str + path_prefix: bool = False + + def guarded(call): + try: + return call() + except KeyError: + raise HTTPException(404, 'Not found') + except ApprovalConflict as exc: + raise HTTPException(409, str(exc)) + except (ValueError, PermissionError) as exc: + raise HTTPException(400, str(exc)) + + @router.get('/approvals') + def approvals(auth=Depends(authenticate)): + def read(): + owner, _ = principal(auth) + service.recover_rules() + return {'requests': [public_record(row) for row in service.approvals.list(owner=owner)]} + return guarded(read) + + @router.get('/approvals/{approval_id}') + def approval(approval_id: str, auth=Depends(authenticate)): + return guarded(lambda: public_record(service.approvals.get(approval_id, owner=principal(auth)[0]))) + + @router.post('/approvals/{approval_id}/decision') + def decide(approval_id: str, body: Decision, auth=Depends(authenticate)): + return guarded(lambda: service.decide(approval_id, auth, body.decision, body.fingerprint)) + + @router.get('/events') + def events(after: int = Query(0, ge=0), auth=Depends(authenticate)): + def read(): + owner, _ = principal(auth) + service.approvals.list(owner=owner) # expiry produces durable events + rows = service.approvals.events(owner=owner, after=after) + return {'events': rows, 'cursor': rows[-1]['sequence'] if rows else after} + return guarded(read) + + @router.get('/rules') + def rules(auth=Depends(authenticate)): + def read(): + principal(auth) + data = service.policy.load() + return {**data, 'rules': [asdict(rule) for rule in data['rules']]} + return guarded(read) + + @router.post('/rules') + def add_rule(body: RuleInput, auth=Depends(authenticate)): + def add(): + _, actor = principal(auth) + if body.decision == 'allow' and (body.operation not in _KNOWN or body.operation in _OPAQUE): + raise ValueError('Opaque or unknown actions cannot receive permanent grants') + return asdict(service.policy.add(actor=actor, approval_id='manual-rule', **body.model_dump())) + return guarded(add) + + @router.delete('/rules/{rule_id}') + def revoke_rule(rule_id: str, auth=Depends(authenticate)): + return guarded(lambda: asdict(service.policy.revoke(rule_id, actor=principal(auth)[1]))) + + return router diff --git a/docs/plans/always-on-policy.md b/docs/plans/always-on-policy.md index 3a99999b..d5a46e35 100644 --- a/docs/plans/always-on-policy.md +++ b/docs/plans/always-on-policy.md @@ -23,3 +23,26 @@ ApprovalStore is an isolated SQLite backend. Requests bind an owner, responsibil A claimed request cannot be automatically replayed after a restart. Claiming is only a reservation; it does not prove a side effect happened and does not replace current policy authorization. A real executor must revalidate identity, policy, scope and request state at the side-effect boundary, record completion/uncertain outcomes, and reconcile an ambiguous external result before any retry. Next bounded slice: authenticated owner/approver mapping, request lists and sanitized immutable review previews, reconnectable events, bounded always-allow rule transactions and process-safe coordination with JSON policies. Always-allow is explicitly rejected by the current storage interface until that transaction exists. A transactional outbox/recovery protocol is needed across SQLite and JSON; do not treat separate DB and JSON writes as atomic. Expiry currently advances when a request is accessed; the API/coordinator must sweep and notify waiting requests. Retention controls and schema migration paths are required before deployment. No public endpoints, live UI delivery or runtime wiring are enabled yet. + +## Shared approval contract (#522) +`/api/v1/autonomy/approvals`, `/approvals/{id}`, `/approvals/{id}/decision`, +`/events?after=N`, and `/rules` require the existing API bearer authentication. +All validated paired/session and shared-key clients belong to this API account; +header-supplied identities cannot grant authority or forge the audited actor. +Events support durable cursor replay; clients refresh authoritative requests on +reconnect. Clients must submit the immutable fingerprint they reviewed. + +Always-allow is an exact agent/operation/host/resource grant. Opaque/unknown +operations cannot get permanent grants. A SQLite outbox records the winning +intent; JSON publication uses private process-safe locks and idempotent IDs. +Execution remains blocked while publication is incomplete. Reconciliation never +reactivates a revoked grant. Rule management supports bounded explicit grants +and revocation. Runtime state defaults to ~/.local/state/wee/autonomy, or the +private WEE_ALWAYS_ON_STATE_DIR; no credentials belong in these files. + +The sole Always-On execution boundary revalidates policy and reserves every +intent before calling a trusted structured adapter. Side-effect failures and +interrupted claims are uncertain and cannot be automatically replayed. No +regular chat/CLI/SDK/browser/delegation adapter is enabled for Always-On. The +feature remains disabled until the coordinator supplies a restricted adapter +registry and opt-in responsibilities. This is not yet a running Always-On agent. diff --git a/tests/test_issue_520_autonomy_policy.py b/tests/test_issue_520_autonomy_policy.py index 8de881e5..4605b2b0 100644 --- a/tests/test_issue_520_autonomy_policy.py +++ b/tests/test_issue_520_autonomy_policy.py @@ -68,7 +68,7 @@ def fail(*args): raise OSError('disk failure') monkeypatch.setattr('autonomy_policy.os.replace', fail) with pytest.raises(OSError): store.add(**options) assert store.path.read_bytes() == previous - assert len(list(tmp_path.iterdir())) == 1 + assert {item.name for item in tmp_path.iterdir()} == {'policy.json', 'policy.json.lock'} @pytest.mark.parametrize('changes', [dict(host='*'), dict(agent='*'), dict(resource='*'), dict(path_prefix=True), dict(enabled='true')]) diff --git a/tests/test_issue_522_autonomy_service.py b/tests/test_issue_522_autonomy_service.py new file mode 100644 index 00000000..308d308c --- /dev/null +++ b/tests/test_issue_522_autonomy_service.py @@ -0,0 +1,94 @@ +from concurrent.futures import ThreadPoolExecutor +import json +import pytest +from autonomy_policy import Action, PolicyStore +from autonomy_approvals import ApprovalConflict +from autonomy_service import ApprovalService, OWNER, principal, create_router + +AUTH = {'auth_type': 'shared_key', 'identity': 'forged-header'} + +def request(service, key='intent'): + return service.approvals.create(Action('wee-dev', 'service.restart', 'dev', 'wee'), owner=OWNER, responsibility='health', intent_key=key, preview={'summary':'Restart Wee on dev'}) + + +def test_identity_is_server_verified(): + assert principal(AUTH) == (OWNER, 'shared-key-client') + assert principal({'auth_type':'session_token', 'identity':'ios'}) == (OWNER,'paired:ios') + with pytest.raises(PermissionError): principal({'identity':'admin'}) + + +def test_rule_publication_recovers_without_duplicate_or_reactivation(tmp_path): + s=ApprovalService(tmp_path); s.policy.set_enabled(True); r=request(s) + row,won=s.approvals.resolve_always(r['id'],owner=OWNER,actor='web',fingerprint=r['fingerprint']) + assert won and row['status']=='rule_pending' + assert not s.approvals.claim(r['id'],Action('wee-dev','service.restart','dev','wee'),owner=OWNER,responsibility='health') + # Simulate JSON committed, SQLite publication uncommitted after a crash. + with s.approvals._transaction() as db: + out=dict(db.execute('SELECT * FROM approval_rule_outbox').fetchone()) + rule=s.policy.add(actor='web',approval_id=r['id'],decision='allow',rule_id=out['rule_id'],**json.loads(out['scope_json'])) + s.policy.revoke(rule.id,actor='mac') + ApprovalService(tmp_path).recover_rules() + assert len(s.policy.load()['rules'])==1 + assert not s.policy.load()['rules'][0].enabled + assert s.approvals.get(r['id'],owner=OWNER)['status']=='expired' + + +def test_three_clients_always_allow_first_wins(tmp_path): + s=ApprovalService(tmp_path);s.policy.set_enabled(True);r=request(s) + with ThreadPoolExecutor(max_workers=3) as pool: + results=list(pool.map(lambda actor: ApprovalService(tmp_path).decide(r['id'],{'auth_type':'session_token','identity':actor},'approve_always',r['fingerprint']),['mac','ios','web'])) + assert sum(result['won'] for result in results)==1 + assert len(s.policy.load()['rules'])==1 + assert all(result['request']['status']=='approved' for result in results) + + +def test_gate_revalidates_deny_and_reserves_every_execution(tmp_path): + s=ApprovalService(tmp_path);s.policy.set_enabled(True);r=request(s) + s.decide(r['id'],AUTH,'approve_once',r['fingerprint']) + s.policy.add(actor='web',approval_id='manual',agent='wee-dev',operation='service.restart',host='dev',resource='wee',decision='deny') + calls=[];a=Action('wee-dev','service.restart','dev','wee') + kwargs=dict(responsibility='health',intent_key='intent',summary='Restart',adapter=lambda a:calls.append(a)) + assert s.execute(a,**kwargs)['status']=='denied' and not calls + rule=s.policy.load()['rules'][0];s.policy.revoke(rule.id,actor='mac') + assert s.execute(a,**kwargs)['status']=='succeeded' + assert s.execute(a,**kwargs)['status']=='succeeded' and len(calls)==1 + with pytest.raises(PermissionError):s.execute(Action('wee-dev','shell.execute','dev','shell'),**kwargs) + + +def test_external_failure_is_uncertain_and_not_replayed(tmp_path): + s=ApprovalService(tmp_path);s.policy.set_enabled(True);r=request(s);s.decide(r['id'],AUTH,'approve_once',r['fingerprint']) + calls=[] + def fail(a): calls.append(1); raise RuntimeError('after external effect') + kwargs=dict(responsibility='health',intent_key='intent',summary='Restart',adapter=fail) + with pytest.raises(RuntimeError):s.execute(Action('wee-dev','service.restart','dev','wee'),**kwargs) + assert s.execute(Action('wee-dev','service.restart','dev','wee'),**kwargs)['status']=='uncertain' + assert len(calls)==1 + + +def test_api_auth_all_shared_clients_and_no_arguments_in_preview(tmp_path): + from fastapi import FastAPI, HTTPException, Header + from fastapi.testclient import TestClient + async def auth(authorization: str=Header('')): + if authorization!='Bearer test':raise HTTPException(401) + return AUTH + s=ApprovalService(tmp_path);s.policy.set_enabled(True);r=request(s) + app=FastAPI();app.include_router(create_router(s,auth));client=TestClient(app) + assert client.get('/api/v1/autonomy/approvals').status_code==401 + h={'Authorization':'Bearer test'} + response=client.get('/api/v1/autonomy/approvals',headers=h) + assert response.status_code==200 and response.json()['requests'][0]['preview']['summary']=='Restart Wee on dev' + for forbidden in ('arguments_json','intent_key','owner'): + assert forbidden not in response.text + assert client.post('/api/v1/autonomy/approvals/'+r['id']+'/decision',headers=h,json={'decision':'approve_always','fingerprint':r['fingerprint']}).status_code==200 + assert len(client.get('/api/v1/autonomy/rules',headers=h).json()['rules'])==1 + cursor=client.get('/api/v1/autonomy/events',headers=h).json()['cursor'] + assert client.get('/api/v1/autonomy/events',params={'after':cursor},headers=h).json()['events']==[] + + +def test_process_safe_writers_do_not_lose_rules(tmp_path): + from multiprocessing import get_context + path=tmp_path/'rules.json' + # Separate store instances exercise independent advisory locks. + with ThreadPoolExecutor(max_workers=12) as pool: + list(pool.map(lambda i: PolicyStore(path).add(actor='user',approval_id=str(i),agent='a',operation='file.read',host='dev',resource='/tmp/'+str(i),decision='allow'),range(40))) + assert len(PolicyStore(path).load()['rules'])==40