local identity

Author: rcholic

README.md

@@ -171,6 +171,46 @@ predicate-authorityd \
   --local-idp-audience "api://predicate-authority"
 ```
 
+### Local identity registry (ephemeral + TTL + flush queue)
+
+Enable sidecar-managed local task identities and local ledger queue:
+
+```bash
+PYTHONPATH=. predicate-authorityd \
+  --host 127.0.0.1 \
+  --port 8787 \
+  --mode local_only \
+  --policy-file examples/authorityd/policy.json \
+  --identity-mode local-idp \
+  --local-identity-enabled \
+  --local-identity-registry-file ./.predicate-authorityd/local-identities.json \
+  --local-identity-default-ttl-s 900 \
+  --flush-worker-enabled \
+  --flush-worker-interval-s 2.0 \
+  --flush-worker-max-batch-size 50 \
+  --flush-worker-dead-letter-max-attempts 5
+```
+
+Runtime endpoints:
+
+- `POST /identity/task` (issue ephemeral task identity)
+- `GET /identity/list` (list identities)
+- `POST /identity/revoke` (revoke identity)
+- `GET /ledger/flush-queue` (inspect pending local ledger queue)
+- `GET /ledger/dead-letter` (list quarantined queue items only)
+- `POST /ledger/flush-ack` (mark queue item as flushed)
+- `POST /ledger/flush-now` (manually trigger immediate queue flush)
+- `POST /ledger/requeue` (requeue quarantined item for retry)
+
+Background flush worker status fields:
+
+- `flush_cycle_count`
+- `flush_sent_count`
+- `flush_failed_count`
+- `flush_quarantined_count`
+- `last_flush_epoch_s`
+- `last_flush_error`
+
 ### How to run with control-plane shipping (out-of-the-box)
 
 ```bash

docs/authorityd-operations.md

@@ -69,6 +69,69 @@ authority decision pushes:
 - audit events -> `/v1/audit/events:batch`
 - usage credits -> `/v1/metering/usage:batch`
 
+## 3b) Optional local identity registry (ephemeral task identities)
+
+Enable local identity support:
+
+```bash
+PYTHONPATH=. predicate-authorityd \
+  --host 127.0.0.1 \
+  --port 8787 \
+  --mode local_only \
+  --policy-file examples/authorityd/policy.json \
+  --identity-mode local-idp \
+  --local-identity-enabled \
+  --local-identity-registry-file ./.predicate-authorityd/local-identities.json \
+  --local-identity-default-ttl-s 900 \
+  --flush-worker-enabled \
+  --flush-worker-interval-s 2.0 \
+  --flush-worker-max-batch-size 50 \
+  --flush-worker-dead-letter-max-attempts 5
+```
+
+Issue an ephemeral identity:
+
+```bash
+curl -s -X POST http://127.0.0.1:8787/identity/task \
+  -H "Content-Type: application/json" \
+  -d '{"principal_id":"agent:backend","task_id":"refactor-pr-102","ttl_seconds":120}'
+```
+
+Inspect pending local ledger flush queue:
+
+```bash
+curl -s http://127.0.0.1:8787/ledger/flush-queue | jq
+```
+
+List quarantined dead-letter items only:
+
+```bash
+curl -s http://127.0.0.1:8787/ledger/dead-letter | jq
+```
+
+Manually trigger an immediate flush cycle:
+
+```bash
+curl -s -X POST http://127.0.0.1:8787/ledger/flush-now \
+  -H "Content-Type: application/json" \
+  -d '{"max_items":50}' | jq
+```
+
+Requeue a quarantined item for retry:
+
+```bash
+curl -s -X POST http://127.0.0.1:8787/ledger/requeue \
+  -H "Content-Type: application/json" \
+  -d '{"queue_item_id":"q_abc123"}' | jq
+```
+
+Flush worker behavior:
+
+- reuses control-plane client retry policy (`--control-plane-max-retries`, `--control-plane-backoff-initial-s`),
+- drains up to `--flush-worker-max-batch-size` queue items per cycle,
+- quarantines entries after `--flush-worker-dead-letter-max-attempts` failed sends,
+- sleeps `--flush-worker-interval-s` between flush cycles.
+
 Expected startup output:
 
 ```text

predicate_authority/README.md

@@ -9,4 +9,5 @@ Core pieces:
 - `LocalMandateSigner` for signed short-lived mandates,
 - `InMemoryProofLedger` and optional `OpenTelemetryTraceEmitter`,
 - typed integration adapters (including `sdk-python` mapping helpers),
-- control-plane client primitives for shipping proof and usage batches to hosted APIs.
+- control-plane client primitives for shipping proof and usage batches to hosted APIs,
+- local identity registry primitives (ephemeral task identities + local flush queue).

predicate_authority/__init__.py

@@ -19,6 +19,14 @@
 from predicate_authority.daemon import DaemonConfig, PredicateAuthorityDaemon
 from predicate_authority.errors import AuthorizationDeniedError
 from predicate_authority.guard import ActionExecutionResult, ActionGuard
+from predicate_authority.local_identity import (
+    CompositeTraceEmitter,
+    LedgerQueueItem,
+    LocalIdentityRegistry,
+    LocalIdentityRegistryStats,
+    LocalLedgerQueueEmitter,
+    TaskIdentityRecord,
+)
 from predicate_authority.mandate import LocalMandateSigner
 from predicate_authority.policy import PolicyEngine, PolicyMatchResult
 from predicate_authority.policy_source import PolicyFileSource, PolicyReloadResult
@@ -53,6 +61,9 @@
     "LocalIdPBridge",
     "LocalIdPBridgeConfig",
     "LocalCredentialStore",
+    "LocalIdentityRegistry",
+    "LocalIdentityRegistryStats",
+    "LocalLedgerQueueEmitter",
     "LocalMandateSigner",
     "LocalRevocationCache",
     "OIDCBridgeConfig",
@@ -68,5 +79,8 @@
     "SidecarError",
     "SidecarStatus",
     "TokenExchangeResult",
+    "CompositeTraceEmitter",
+    "LedgerQueueItem",
+    "TaskIdentityRecord",
     "UsageCreditRecord",
 ]

predicate_authority/control_plane.py

@@ -97,6 +97,9 @@ def send_usage_records(self, records: tuple[UsageCreditRecord, ...]) -> bool:
         payload = {"records": [asdict(record) for record in records]}
         return self._post_json("/v1/metering/usage:batch", payload)
 
+    def send_audit_payload(self, payload: Mapping[str, object]) -> bool:
+        return self._post_json("/v1/audit/events:batch", payload)
+
     def _post_json(self, path: str, payload: Mapping[str, object]) -> bool:
         attempts = self.config.max_retries + 1
         for attempt in range(attempts):