fix(runtime): isolate decision observation failures

Defer terminal decision events until the Atomic Memory transaction commits, preserve shadow writes when audit persistence fails, and disable redirects for custom Anthropic header clients.

Constraint: audit writes stay in the owner transaction and malformed audit writes never block the domain write.

Scope-risk: changes are limited to decision observation finalization and custom provider redirect transport.

Tested: uv run --no-sync pytest tests/builtin/runtime/test_decision_composition.py tests/builtin/runtime/test_memory_window_recovery.py -q; uv run --no-sync pytest tests/builtin/runtime/test_memory_write_gate_paths.py tests/builtin/runtime/test_processing_composition.py tests/builtin/runtime/test_memory_write_gate_contract.py -q; ruff and ty checks passed.

Author: AlexStocks

src/powercontext/builtin/runtime/atomic_memory_processing.py

@@ -16,6 +16,7 @@
 
 from __future__ import annotations
 
+import logging
 import math
 from base64 import urlsafe_b64encode
 from contextlib import asynccontextmanager, nullcontext
@@ -25,6 +26,7 @@
 
 from sqlalchemy import update
 
+from powercontext._logging import log_safely
 from powercontext.artifacts import ArtifactLineage, ArtifactRef
 from powercontext.artifacts.search import ArtifactSearchExecutionContext
 from powercontext.builtin.artifacts.atomic_memory.errors import AtomicMemoryConflictError
@@ -77,6 +79,9 @@
     from powercontext.builtin.runtime.config import RuntimeConfig
 
 
+logger = logging.getLogger(__name__)
+
+
 @dataclass(frozen=True)
 class AtomicMemoryProcessingConfig:
     related_mode: Literal["auto", "fts", "vector", "hybrid"] = "auto"
@@ -215,22 +220,28 @@ async def _flush_once(self, scope_id, limit, processing, context, span) -> Memor
             if reducible and action.through - action.after > 1:
                 await self._reduce_window(scope_id, context, processing, cursor, state, action, high_watermark)
             raise
-        await self._publish(
-            scope_id,
-            context,
-            processing,
-            cursor,
-            transition,
-            action,
-            window,
-            eligible,
-            consulted_artifacts,
-            workset,
-            prepared,
-            decision,
-            observation,
-            high_watermark,
-        )
+        try:
+            finalized = await self._publish(
+                scope_id,
+                context,
+                processing,
+                cursor,
+                transition,
+                action,
+                window,
+                eligible,
+                consulted_artifacts,
+                workset,
+                prepared,
+                decision,
+                observation,
+                high_watermark,
+            )
+        except Exception:
+            self._emit_failed_observation(observation)
+            raise
+        if finalized is not None:
+            emit_decision_observation(finalized)
         return self._result(
             action.after,
             action.through,
@@ -310,6 +321,7 @@ async def _publish(
     ):
         application = self.application
         writes = tuple(write for item in prepared for write in item.plan.writes)
+        finalized = None
         with self._stage(
             "memory.commit",
             {
@@ -352,7 +364,7 @@ async def _publish(
                     results = await application.service.commit_window(
                         connection, scope_id, prepared, context, read_set=tuple(workset.reads.values())
                     )
-                await self._finalize_observation(
+                finalized = await self._finalize_observation(
                     connection,
                     observation,
                     final_action=(
@@ -372,6 +384,7 @@ async def _publish(
                 await self.windows.clear_consumed(connection, scope_id, action.through)
                 if processing is not None:
                     await processing.complete(connection, remaining_work=action.through < high_watermark)
+        return finalized
 
     async def _prepare(self, scope_id, eligible, pipeline, context):
         workset = AtomicMemoryWindowWorkset()
@@ -638,23 +651,41 @@ async def _assess_write_gate(self, scope_id, workset, prepared):
 
     async def _finalize_observation(self, connection, observation, *, final_action, subject_refs):
         if observation is None or self.decision_observations is None:
-            return
+            return None
         finalized = observation.model_copy(
             update={
                 "final_action": final_action,
                 "subject_refs": observation.subject_refs if final_action == "memory_write_hold" else subject_refs,
             }
         )
-        await self.decision_observations.append(connection, finalized)
-        if self.decision_observation_retention_days is not None:
-            from datetime import UTC, datetime, timedelta
+        try:
+            async with connection.begin_nested():
+                await self.decision_observations.append(connection, finalized)
+                if self.decision_observation_retention_days is not None:
+                    from datetime import UTC, datetime, timedelta
 
-            await self.decision_observations.prune_before(
-                connection,
-                finalized.scope_id,
-                datetime.now(UTC) - timedelta(days=self.decision_observation_retention_days),
+                    await self.decision_observations.prune_before(
+                        connection,
+                        finalized.scope_id,
+                        datetime.now(UTC) - timedelta(days=self.decision_observation_retention_days),
+                    )
+        except Exception as error:
+            log_safely(
+                logger,
+                logging.WARNING,
+                "Decision observation persistence failed",
+                exc_info=error,
+                extra={"event": "decision.observation_sink_failed", "error_type": type(error).__name__},
+            )
+            return None
+        return finalized
+
+    @staticmethod
+    def _emit_failed_observation(observation):
+        if observation is not None:
+            emit_decision_observation(
+                observation.model_copy(update={"final_action": "memory_write_failed", "subject_refs": ()})
             )
-        emit_decision_observation(finalized)
 
     @staticmethod
     def _result(after, through, high_watermark, count, *, held=False, hold_codes=()):

src/powercontext/builtin/runtime/composition.py

@@ -1733,7 +1733,7 @@ def create_provider(provider_name: str) -> Provider[Any]:
             if headers:
                 from openai import AsyncOpenAI
 
-                http_client = _openai_http_client(follow_redirects=False) if base_url is not None else None
+                http_client = _endpoint_http_client(follow_redirects=False) if base_url is not None else None
                 client = AsyncOpenAI(
                     base_url=None if base_url is None else str(base_url),
                     api_key=os.getenv("OPENAI_API_KEY") or "api-key-not-set",
@@ -1748,7 +1748,7 @@ def create_provider(provider_name: str) -> Provider[Any]:
                 client = AsyncOpenAI(
                     base_url=str(base_url),
                     api_key=os.getenv("OPENAI_API_KEY") or "api-key-not-set",
-                    http_client=_openai_http_client(follow_redirects=False),
+                    http_client=_endpoint_http_client(follow_redirects=False),
                 )
                 resources.push_async_callback(client.close)
                 return OpenAIProvider(openai_client=client)
@@ -1764,6 +1764,7 @@ def create_provider(provider_name: str) -> Provider[Any]:
                     base_url=None if base_url is None else str(base_url),
                     api_key=api_key or os.getenv("ANTHROPIC_API_KEY") or "api-key-not-set",
                     default_headers=default_headers,
+                    http_client=_endpoint_http_client(follow_redirects=False) if base_url is not None else None,
                 )
                 resources.push_async_callback(client.close)
                 return AnthropicProvider(anthropic_client=client)
@@ -1776,8 +1777,8 @@ def create_provider(provider_name: str) -> Provider[Any]:
     return create_provider
 
 
-def _openai_http_client(*, follow_redirects: bool):
-    """Create the SDK transport with an explicit redirect policy for local-only endpoints."""
+def _endpoint_http_client(*, follow_redirects: bool):
+    """Create the SDK transport with an explicit redirect policy for custom endpoints."""
 
     import httpx
 

tests/builtin/runtime/test_decision_composition.py

@@ -20,7 +20,7 @@
 from typing import Any
 
 import pytest
-from pydantic import AnyHttpUrl
+from pydantic import AnyHttpUrl, SecretStr
 
 from powercontext.builtin.inference import InferenceUnavailableError, InferenceUsage
 from powercontext.builtin.persistence.sqlite import SQLiteConfig
@@ -231,3 +231,20 @@ async def scenario() -> None:
             assert provider.client._client.follow_redirects is False  # pyright: ignore[reportPrivateUsage]
 
     asyncio.run(scenario())
+
+
+def test_loopback_anthropic_headers_provider_disables_redirects() -> None:
+    async def scenario() -> None:
+        async with AsyncExitStack() as resources:
+            factory = composition._unbounded_provider_factory(
+                AnyHttpUrl("http://127.0.0.1:11434"),
+                {"x-api-key": SecretStr("test-key")},
+                workload="generation",
+                resources=resources,
+            )
+
+            provider = factory("anthropic")
+
+            assert provider.client._client.follow_redirects is False  # pyright: ignore[reportPrivateUsage]
+
+    asyncio.run(scenario())

tests/builtin/runtime/test_memory_window_recovery.py

@@ -18,6 +18,8 @@
 from uuid import uuid4
 
 import pytest
+from sqlalchemy import text
+from sqlalchemy.exc import IntegrityError
 
 from powercontext.builtin.artifacts.atomic_memory.extraction import AtomicMemoryCandidate, AtomicMemoryExtractionOutput
 from powercontext.builtin.decision_observations import DecisionPrivacyBoundary
@@ -165,6 +167,87 @@ async def scenario():
     assert [record.__dict__["final_action"] for record in events] == ["memory_write_applied"]
 
 
+def test_atomic_memory_gate_emits_failed_outcome_after_late_transaction_rollback(
+    tmp_path, caplog: pytest.LogCaptureFixture
+):
+    async def scenario():
+        gate = build_memory_write_gate(
+            None,
+            enabled=True,
+            privacy_boundary=DecisionPrivacyBoundary.NO_EXTERNAL_CALL,
+        )
+        assert gate is not None
+        config = SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'atomic-gate-rollback.db'}")
+        async with SQLiteProfile.open(config, tables=BUILTIN_TABLES) as profile:
+            contexts = await _ready_contexts(
+                profile,
+                candidate_pipeline=atomic_memory_pipeline(Pipeline()),
+                memory_write_gate=gate,
+            )
+            scope = await create_scope(contexts, 1)
+            context = await contexts.get(scope)
+            async with profile.database.transaction() as connection:
+                await connection.execute(
+                    text(
+                        "CREATE TRIGGER reject_memory_cursor "
+                        "BEFORE INSERT ON pc_source_cursors "
+                        "BEGIN SELECT RAISE(ABORT, 'cursor rejected'); END"
+                    )
+                )
+
+            with pytest.raises(IntegrityError, match="cursor rejected"):
+                await context.triggers.flush(limit=1)
+
+            assert (await context.triggers.cursor()).sequence == 0
+            assert not (await contexts.atomic_memory.for_scope(scope).list()).items
+            async with profile.database.transaction() as connection:
+                assert not await DecisionObservationRepository().observations(connection, scope)
+
+    caplog.set_level("INFO", logger="powercontext.builtin.runtime.decision_policy")
+    asyncio.run(scenario())
+    events = [record for record in caplog.records if getattr(record, "event", None) == "decision.observation"]
+    assert [record.__dict__["final_action"] for record in events] == ["memory_write_failed"]
+
+
+def test_atomic_memory_audit_failure_does_not_block_shadow_write(tmp_path, caplog: pytest.LogCaptureFixture):
+    async def scenario():
+        gate = build_memory_write_gate(
+            None,
+            enabled=True,
+            privacy_boundary=DecisionPrivacyBoundary.NO_EXTERNAL_CALL,
+        )
+        assert gate is not None
+        config = SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'atomic-gate-audit.db'}")
+        async with SQLiteProfile.open(config, tables=BUILTIN_TABLES) as profile:
+            contexts = await _ready_contexts(
+                profile,
+                candidate_pipeline=atomic_memory_pipeline(Pipeline()),
+                memory_write_gate=gate,
+            )
+            scope = await create_scope(contexts, 1)
+            context = await contexts.get(scope)
+            async with profile.database.transaction() as connection:
+                await connection.execute(
+                    text(
+                        "CREATE TRIGGER reject_decision_observation "
+                        "BEFORE INSERT ON pc_decision_observations "
+                        "BEGIN SELECT RAISE(ABORT, 'audit rejected'); END"
+                    )
+                )
+
+            assert (await context.triggers.flush(limit=1)).current_cursor == 1
+            assert len((await contexts.atomic_memory.for_scope(scope).list()).items) == 1
+            async with profile.database.transaction() as connection:
+                assert not await DecisionObservationRepository().observations(connection, scope)
+
+    caplog.set_level("WARNING", logger="powercontext.builtin.runtime.atomic_memory_processing")
+    asyncio.run(scenario())
+    diagnostics = [
+        record for record in caplog.records if getattr(record, "event", None) == "decision.observation_sink_failed"
+    ]
+    assert len(diagnostics) == 1
+
+
 @pytest.mark.parametrize("error", [InferenceTimeoutError("embed", 60), InferenceUnavailableError("generate")])
 def test_non_extraction_failures_do_not_reduce_source_window(error):
     async def scenario():