Implement committed native effects and guarded worker results

.github/workflows/test.yml

@@ -53,7 +53,7 @@ jobs:
         uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
         with:
           repository: fruwehq/determa-state-conformance
-          ref: ce6c94ee1929597689403080d1790e2f48f8a5fb
+          ref: 7f09321fb483a22eb677a4342f8d9537a7a18e82
           path: .pinned/determa-state-conformance
       - name: Check out pinned specification
         uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
@@ -88,7 +88,7 @@ jobs:
         uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
         with:
           repository: fruwehq/determa-state-conformance
-          ref: ce6c94ee1929597689403080d1790e2f48f8a5fb
+          ref: 7f09321fb483a22eb677a4342f8d9537a7a18e82
           path: .pinned/determa-state-conformance
       - name: Check out pinned specification
         uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
@@ -126,7 +126,7 @@ jobs:
         uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
         with:
           repository: fruwehq/determa-state-conformance
-          ref: ce6c94ee1929597689403080d1790e2f48f8a5fb
+          ref: 7f09321fb483a22eb677a4342f8d9537a7a18e82
           path: .pinned/determa-state-conformance
       - name: Check out pinned specification
         uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
@@ -151,6 +151,10 @@ jobs:
           PYTHONPATH=src python "$DETERMA_CONFORMANCE_DIR/scripts/run_host_authority_profile.py" \
             --spec-root "$DETERMA_SPEC_DIR" \
             --adapter python conformance/authority_adapter.py
+          PYTHONPATH=src:. python "$DETERMA_CONFORMANCE_DIR/scripts/run_committed_native_effects_profile.py" \
+            --spec-root "$DETERMA_SPEC_DIR" \
+            --adapter "python conformance/effects_adapter.py" \
+            --authority-adapter "python conformance/worker_authority_adapter.py"
 
   postgresql:
     runs-on: ubuntu-24.04

AGENTS.md

@@ -12,10 +12,10 @@ The implementation is conformant only when it passes the language-neutral suite.
 The implementation uses these immutable inputs:
 
 - specification: `86bb88dd21cb1f799eefe5020b6e49dabf6e7225`;
-- conformance: `ce6c94ee1929597689403080d1790e2f48f8a5fb` (99 format-1
+- conformance: `7f09321fb483a22eb677a4342f8d9537a7a18e82` (99 format-1
   core cases, 162 version-1 vectors, 142 durable-host vectors, 49 core inspection
   vectors, 47 extension negotiation vectors, 7 optional native-provider vectors,
-  30 runtime-provider vectors, and 476 JSON artifacts).
+  30 runtime-provider vectors, and 484 JSON artifacts).
 
 The package metadata is `0.3.0`.
 

README.md

@@ -7,8 +7,8 @@ This implementation supports Determa State `format: 1` at specification
 commit `86bb88dd21cb1f799eefe5020b6e49dabf6e7225`. Correctness is determined by
 99 format-1 core cases, 162 version-1 vectors, 142 durable-host vectors, 49 core
 inspection vectors, 47 extension negotiation vectors, 7 native guard-provider
-inspection vectors, 30 runtime-provider vectors, and 476 manifest artifacts at
-conformance commit `ce6c94ee1929597689403080d1790e2f48f8a5fb`. The native
+inspection vectors, 30 runtime-provider vectors, and 484 manifest artifacts at
+conformance commit `7f09321fb483a22eb677a4342f8d9537a7a18e82`. The native
 runtime-provider profile is optional and is exercised through its production adapter.
 
 The package metadata is `0.3.0`. Artifact and checkpoint schema version 1 is the
@@ -391,11 +391,48 @@ health and the same store instance used by the host. Recheck it before relying o
 the guarded profile. A frozen scope retains fencing and inventory claims while
 checkpoint writes remain refused.
 
-The reference boundary does not issue worker claims or retirement grants and
-reports `worker_fencing: false` and `safe_relocation: false`. A copied database,
+The default checkpoint-only configuration reports `worker_fencing: false`.
+An explicitly configured colocated worker journal can issue local claims with
+`worker_fencing=True`; it binds the root, effect identity and operation token
+before issuance, and commits journal state, claim, generation and receipt together.
+The reference topology does not issue retirement grants and reports
+`safe_relocation: false`. A copied database,
 new endpoint, or timeout cannot activate another owner. The pure in-memory API
 needs no scope authority.
 
+## Optional committed native effects
+
+`determa.state.effects.SQLiteCommittedEffectHost` implements the optional local
+SQLite effect journal. Install a `VerifiedNativeHandler` through the ordinary
+`ExtensionRegistry` with a trusted source verifier, exact provider reference,
+configured instance and destination binding. Plain callbacks cannot be installed
+as handlers. A host without a handler may restore or reconcile committed data;
+producing or dispatching external work requires the verified installed handle.
+SDK objects can remain private inside the handler; boundary payloads use declared
+portable typed values.
+
+The host commits a pinned effect intent before dispatch. Worker calls use an
+independently authenticated principal, current scope/epoch/attempt claim and an
+explicit `trusted_clock` callback. Clock failures or expiry fail closed, including
+expiry during outcome recording before its native commit. An expired external
+attempt remains unresolved and requires destination idempotency or reconciliation;
+lease expiry does not prove the call was undone. An ambiguous retry requires
+`deduplication_evidence` bound to the scope, effect and destination. The installed
+provider's `verify_deduplication_evidence(instance, evidence)` must authenticate
+both receipt byte strings against its actual native destination; equal bytes from
+a caller are insufficient. The host retains the verified evidence with the new
+attempt fence in the same transaction as the claim.
+
+The public `submit_result` returns `committed` only after result-event admission.
+Recovery verifies the exact checkpoint acceptance receipt against the pinned
+payload, token mapping and original target identity, even after processing or
+root tombstoning. Authenticated outcome recording commits before result-event admission. A crash
+between those transactions retains the outcome, and `recover` admits it without
+calling the provider again. Each recoverable effect commits separately. Duplicate
+results retain exact first responses; conflicting or stale work is refused.
+Delivery acknowledgement and business outcome remain separate. The optional host
+does not create a background worker or claim arbitrary external exactly-once work.
+
 ## Develop
 
 ```sh

conformance/authority_adapter.py

@@ -22,18 +22,26 @@
     _parse,
     configure_bundled_sqlite_authority,
 )
+from determa.state.effects import SQLiteCommittedEffectHost
 from determa.state.extensions import bundled_extension_registry
-from determa.state.wire import hash_value
+from determa.state.wire import MemoryArtifactResolver, hash_value
 
 _PATH = Path(tempfile.gettempdir()) / "determa-python-local-authority-profile.sqlite"
 _INITIAL_BINDING: str | None = None
+_WORKER_MODE = "--worker" in sys.argv[1:]
 
 
 def _profile() -> dict[str, Any]:
     authority = _seed(_baseline_ledger())
     registry = bundled_extension_registry(include_postgresql=False)
     configured, authority, store = configure_bundled_sqlite_authority(
-        registry, authority.path, "scope-42", "owner-1", "2"
+        registry,
+        authority.path,
+        "scope-42",
+        "owner-1",
+        "2",
+        worker_fencing=_WORKER_MODE,
+        worker_lease_nanoseconds=1,
     )
     report = authority.profile_report("scope-42", "owner-1", store, registry, configured)
     installed = authority.profile_descriptor(store, registry, configured)["installation_evidence"]
@@ -48,15 +56,34 @@ def _profile() -> dict[str, Any]:
                 "freeze_waits_for_writer",
                 "freeze_after_drain",
                 "incomplete_frozen_inventory_refuses_freeze",
-            ],
+            ]
+            + (
+                [
+                    "fence_worker_allocates_new_claim",
+                    "dispatch_at_expiry",
+                    "result_at_expiry",
+                    "clock_unavailable",
+                    "old_epoch",
+                    "old_attempt",
+                    "principal_mismatch",
+                ]
+                if _WORKER_MODE
+                else []
+            ),
         },
     }
 
 
 def _binding_live() -> tuple[str, SQLiteLocalAuthority]:
     registry = bundled_extension_registry(include_postgresql=False)
     configured, authority, store = configure_bundled_sqlite_authority(
-        registry, _PATH, "scope-42", "owner-1", "2"
+        registry,
+        _PATH,
+        "scope-42",
+        "owner-1",
+        "2",
+        worker_fencing=_WORKER_MODE,
+        worker_lease_nanoseconds=1,
     )
     descriptor = authority.profile_descriptor(store, registry, configured)
     return hash_value(
@@ -112,13 +139,43 @@ def _seed(ledger: dict[str, Any]) -> SQLiteLocalAuthority:
     global _INITIAL_BINDING
     # These trusted protocol fixtures contain opaque mutations, not native rows.
     ledger = {**ledger, "native_checkpoint_bytes": []}
+    if _WORKER_MODE:
+        # Explicit trusted fixture installation, independent of the incoming
+        # request and expected result. These §18 vectors use this fixed work
+        # identity rather than a §19 native checkpoint/effect journal.
+        ledger["authority_effect_bindings"] = [
+            {
+                "root_instance_id": "root-7",
+                "work_kind": "effect",
+                "work_identity": "effect-17",
+                "operation_token": "token-17",
+                "participant": "authority_journal",
+            }
+            for entry in ledger["journal_entries"]
+            if entry["work_identity"] == "effect-17"
+        ]
+        ledger["authority_effect_records"] = [
+            {
+                **entry,
+                "state": "leased"
+                if any(
+                    claim["work_identity"] == entry["work_identity"] and claim["state"] == "active"
+                    for claim in ledger["active_claims"]
+                )
+                else "unclaimed",
+            }
+            for entry in ledger["journal_entries"]
+            if entry["work_identity"] == "effect-17"
+        ]
     # A fresh test database for each independent vector. Production never
     # removes an allocated scope marker from a live authority domain.
     for suffix in ("", "-wal", "-shm"):
         (_PATH.parent / (_PATH.name + suffix)).unlink(missing_ok=True)
-    authority = SQLiteLocalAuthority(_PATH)
+    authority = SQLiteLocalAuthority(_PATH, worker_fencing=_WORKER_MODE, worker_lease_nanoseconds=1)
     authority.setup_schema()
     AuthoritySQLiteExecutionStore(authority, "scope-42", "owner-1", "2").setup_schema()
+    if _WORKER_MODE:
+        SQLiteCommittedEffectHost(_PATH, MemoryArtifactResolver(), {}, None).setup_schema()
     if not authority._insert_ledger(_baseline_ledger()):
         raise RuntimeError("fixture scope already allocated")
     _INITIAL_BINDING, authority = _binding_live()
@@ -138,7 +195,19 @@ def _observed_ledger(authority: SQLiteLocalAuthority) -> dict[str, Any] | None:
     if ledger is None:
         return None
     # Project the actual ledger onto the closed conformance observation shape.
-    return {key: value for key, value in ledger.items() if key != "native_checkpoint_bytes"}
+    return {
+        key: value
+        for key, value in ledger.items()
+        if key
+        not in {
+            "native_checkpoint_bytes",
+            "effect_claim_history",
+            "native_effect_roots",
+            "native_effect_work",
+            "authority_effect_bindings",
+            "authority_effect_records",
+        }
+    }
 
 
 def _call(authority: SQLiteLocalAuthority, call: dict[str, Any], **kwargs: Any) -> str | None:
@@ -187,7 +256,9 @@ def start(session: str, step: int, barrier: str) -> None:
             held = context.Event()
 
             def disconnected_writer() -> None:
-                local = SQLiteLocalAuthority(_PATH)
+                local = SQLiteLocalAuthority(
+                    _PATH, worker_fencing=_WORKER_MODE, worker_lease_nanoseconds=1
+                )
 
                 def hold_transaction() -> None:
                     held.set()
@@ -370,7 +441,25 @@ def run(payload: dict[str, Any]) -> dict[str, Any]:
     if kind == "profile":
         return _common_rule(payload["configured_facts"], set())
     if kind == "worker_claim_check":
-        raise NotImplementedError("worker claim checks unavailable")
+        if not _WORKER_MODE:
+            raise NotImplementedError("worker claim checks unavailable")
+        item = payload["input"]
+        authority = _seed(item["ledger_before"])
+        dispatched: list[str] = []
+        accepted = authority.check_worker_claim(
+            item["claim"],
+            item["authenticated_principal"],
+            item["trusted_clock_now"],
+            phase=item["phase"],
+            on_dispatch=lambda: dispatched.append("called"),
+        )
+        return {
+            "binding": _binding(),
+            "accepted": accepted,
+            "ledger_after": _observed_ledger(authority),
+            "host_mutation_count": 0,
+            "external_dispatch_count": len(dispatched),
+        }
     raise ValueError(f"unknown driver kind: {kind}")