From 04f84fcbad645416a6b16a048de270b66421a33d Mon Sep 17 00:00:00 2001 From: Codex Date: Sun, 23 Aug 2026 22:03:29 +0800 Subject: [PATCH 1/4] feat(feed): hand off legacy reservoir to Content --- content_source.py | 3 + feed_runtime/backend.py | 22 +- legacy_handoff.py | 243 +++++++++++++ plugin.py | 4 +- .../fixtures/legacy_feed_reservoir_rows.json | 17 + tests/test_legacy_handoff.py | 324 ++++++++++++++++++ 6 files changed, 608 insertions(+), 5 deletions(-) create mode 100644 legacy_handoff.py create mode 100644 tests/fixtures/legacy_feed_reservoir_rows.json create mode 100644 tests/test_legacy_handoff.py diff --git a/content_source.py b/content_source.py index eba0116..ac4364b 100644 --- a/content_source.py +++ b/content_source.py @@ -16,6 +16,9 @@ from feed_runtime import backend +CONTENT_SOURCE_ID = "feed-subscriptions" + + class BoundContentSource(Protocol): def submit( self, batch_id: str, items: Sequence[Mapping[str, object]] diff --git a/feed_runtime/backend.py b/feed_runtime/backend.py index e1261b8..9f006e2 100644 --- a/feed_runtime/backend.py +++ b/feed_runtime/backend.py @@ -75,15 +75,31 @@ def _runtime_root(data_root: Path | None = None) -> Path: return path -def load_config(data_root: Path | None = None) -> FeedMcpConfig: - runtime_root = _runtime_root(data_root) +def _config_values() -> dict[str, Any]: raw = dict(_DEFAULT_CONFIG) path = _config_path() if path.exists(): raw.update(json.loads(path.read_text())) + return raw + + +def _database_path(data_root: Path, raw: dict[str, Any]) -> Path: db_path = Path(str(raw["db_path"])) if not db_path.is_absolute(): - db_path = (runtime_root / db_path).resolve() + db_path = (data_root.expanduser() / db_path).resolve() + return db_path + + +def provider_database_path(data_root: Path) -> Path: + """Resolve the configured Feed database without creating runtime state.""" + + return _database_path(data_root, _config_values()) + + +def load_config(data_root: Path | None = None) -> FeedMcpConfig: + runtime_root = _runtime_root(data_root) + raw = _config_values() + db_path = _database_path(runtime_root, raw) return FeedMcpConfig( db_path=db_path, poll_ttl_seconds=max(60, int(raw["poll_ttl_seconds"])), diff --git a/legacy_handoff.py b/legacy_handoff.py new file mode 100644 index 0000000..61e066c --- /dev/null +++ b/legacy_handoff.py @@ -0,0 +1,243 @@ +from __future__ import annotations + +import hashlib +import json +import sqlite3 +from collections.abc import Mapping, Sequence +from contextlib import closing +from pathlib import Path +from typing import Protocol, cast + +from agent.migrations.proactive_island import ( + AdapterPlan, + HandoffBlocked, + LegacyFact, + LegacyFactKind, + TargetReceipt, +) +from agent.migrations.proactive_island.handoff import receipt_digest + +from content_source import CONTENT_SOURCE_ID +from feed_runtime import backend + + +LEGACY_SOURCE_ID = "feed@github:subscriptions" + + +class BoundContentSource(Protocol): + def submit( + self, batch_id: str, items: Sequence[Mapping[str, object]] + ) -> Mapping[str, object]: ... + + def read_submission(self, batch_id: str) -> Mapping[str, object] | None: ... + + def read_revision( + self, item_id: str, revision: str + ) -> Mapping[str, object] | None: ... + + +class FeedLegacyHandoffAdapter: + """Move exact legacy Feed reservoir facts into the existing Content source.""" + + def __init__(self, feed_data_root: Path, content: BoundContentSource) -> None: + self._provider_db = backend.provider_database_path(feed_data_root) + self._content = content + + def accepts(self, fact: LegacyFact) -> bool: + return ( + fact.kind is LegacyFactKind.WAKE_SOURCE_ITEM + and fact.source_identity == LEGACY_SOURCE_ID + ) + + def plan(self, fact: LegacyFact) -> AdapterPlan: + """Resolve one provider-owned revision without mounting or writing Content.""" + + row = _legacy_row(fact) + provider = self._provider_item(_text(row, "source_event_id")) + return AdapterPlan(_target_identity(provider)) + + def apply(self, fact: LegacyFact, plan: AdapterPlan) -> TargetReceipt: + """Submit the exact planned target and return its normalized durable receipt.""" + + # 1. Re-read the owner row and reject a revision change after planning. + row = _legacy_row(fact) + provider = self._provider_item(_text(row, "source_event_id")) + target_identity = _target_identity(provider) + if plan.target_identity != target_identity: + raise RuntimeError("Feed handoff target identity drift after plan") + + # 2. A fact-stable batch makes target-before-marker replay idempotent. + batch_id = _batch_id(fact) + item = _content_item(row, provider) + content_receipt = self._content.submit(batch_id, (item,)) + receipt_id = _text(content_receipt, "receipt_id") + normalized = _receipt_payload(fact, target_identity, content_receipt) + return TargetReceipt( + receipt_id=receipt_id, + receipt_digest=receipt_digest(normalized), + target_identity=target_identity, + ) + + def verify(self, fact: LegacyFact, receipt: TargetReceipt) -> bool: + """Verify provider lineage and checkpointed Content facts without writing.""" + + try: + plan = self.plan(fact) + except HandoffBlocked: + return False + if receipt.target_identity != plan.target_identity: + return False + row = _legacy_row(fact) + provider = self._provider_item(_text(row, "source_event_id")) + item = _content_item(row, provider) + batch_id = _batch_id(fact) + submission = self._content.read_submission(batch_id) + revision = self._content.read_revision( + _text(item, "item_id"), _text(item, "revision") + ) + if submission is None or revision is None: + return False + normalized = _receipt_payload(fact, plan.target_identity, submission) + return ( + receipt.receipt_id == submission.get("receipt_id") + and receipt.receipt_digest == receipt_digest(normalized) + and _revision_matches(revision, item) + ) + + def _provider_item(self, event_id: str) -> dict[str, object]: + """Read one exact Feed row through a query-only SQLite connection.""" + + if not self._provider_db.is_file(): + raise HandoffBlocked("feed_provider_database_missing") + wal = self._provider_db.with_name(self._provider_db.name + "-wal") + if wal.is_file() and wal.stat().st_size > 0: + raise HandoffBlocked("feed_provider_checkpoint_required") + uri = self._provider_db.resolve().as_uri() + "?mode=ro&immutable=1" + with closing(sqlite3.connect(uri, uri=True)) as connection: + connection.row_factory = sqlite3.Row + _ = connection.execute("PRAGMA query_only = ON") + result = connection.execute( + """ + SELECT event_id, source_id, source_type, source_name, title, + content, url, author, published_at, first_seen_at, + content_hash + FROM items WHERE event_id = ? + """, + (event_id,), + ).fetchone() + if result is None: + raise HandoffBlocked(f"feed_provider_item_missing:{event_id}") + return {key: result[key] for key in result.keys()} + + +def _legacy_row(fact: LegacyFact) -> dict[str, object]: + if fact.kind is not LegacyFactKind.WAKE_SOURCE_ITEM: + raise TypeError("Feed handoff received another legacy fact kind") + if fact.source_identity != LEGACY_SOURCE_ID: + raise TypeError("Feed handoff received another legacy source owner") + if hashlib.sha256(fact.opaque).hexdigest() != fact.source_digest: + raise RuntimeError("Feed legacy source digest mismatch") + decoded = json.loads(fact.opaque) + if not isinstance(decoded, dict): + raise TypeError("Feed legacy reservoir row must be an object") + row = cast(dict[str, object], decoded) + event_id = _text(row, "source_event_id") + if not fact.locator.endswith(f":{_text(row, 'item_id')}"): + raise RuntimeError("Feed legacy locator does not match item_id") + payload = _payload(row) + if payload.get("event_id") != event_id or payload.get("kind") != "content": + raise RuntimeError("Feed legacy payload identity mismatch") + return row + + +def _content_item( + legacy: Mapping[str, object], provider: Mapping[str, object] +) -> dict[str, object]: + event_id = _text(provider, "event_id") + if _text(legacy, "source_event_id") != event_id: + raise RuntimeError("Feed provider join identity mismatch") + source_payload = _payload(legacy) + payload: dict[str, object] = { + "kind": "content", + "source_type": provider["source_type"], + "source_id": provider["source_id"], + "source_name": provider["source_name"], + "title": provider["title"], + "content": provider["content"], + "url": provider["url"], + "author": provider["author"], + "published_at": provider["published_at"], + "first_seen_at": provider["first_seen_at"], + "preprocess_score": source_payload.get( + "preprocess_score", legacy["preprocess_score"] + ), + "preprocess_features": source_payload.get("preprocess_features", {}), + } + return { + "item_id": event_id, + "revision": _text(provider, "content_hash"), + "payload": payload, + "not_before": str(provider["published_at"] or provider["first_seen_at"]), + "requires_ack": True, + } + + +def _target_identity(provider: Mapping[str, object]) -> str: + return ( + f"content:{CONTENT_SOURCE_ID}:{_text(provider, 'event_id')}:" + f"{_text(provider, 'content_hash')}" + ) + + +def _batch_id(fact: LegacyFact) -> str: + encoded = f"{fact.locator}\x00{fact.source_digest}".encode("utf-8") + return f"feed-legacy:{hashlib.sha256(encoded).hexdigest()}" + + +def _receipt_payload( + fact: LegacyFact, + target_identity: str, + content_receipt: Mapping[str, object], +) -> dict[str, object]: + return { + "schema_version": 1, + "legacy_locator": fact.locator, + "legacy_source_digest": fact.source_digest, + "legacy_source_identity": fact.source_identity, + "target_identity": target_identity, + "content_receipt": dict(content_receipt), + } + + +def _revision_matches( + revision: Mapping[str, object], item: Mapping[str, object] +) -> bool: + return ( + revision.get("ref") + == { + "source_id": CONTENT_SOURCE_ID, + "item_id": item["item_id"], + "revision": item["revision"], + } + and revision.get("payload") == item["payload"] + and revision.get("not_before") == item["not_before"] + and revision.get("requires_ack") is True + ) + + +def _payload(row: Mapping[str, object]) -> dict[str, object]: + raw = _text(row, "payload_json") + value = json.loads(raw) + if not isinstance(value, dict): + raise TypeError("Feed legacy payload_json must contain an object") + return cast(dict[str, object], value) + + +def _text(row: Mapping[str, object], field: str) -> str: + value = row[field] + if not isinstance(value, str) or not value: + raise TypeError(f"Feed {field} must be a non-empty string") + return value + + +__all__ = ["FeedLegacyHandoffAdapter"] diff --git a/plugin.py b/plugin.py index 01b3fb3..6cf0fee 100644 --- a/plugin.py +++ b/plugin.py @@ -12,7 +12,7 @@ ServiceKey, ) -from content_source import ContentSourceServices, FeedContentRuntime +from content_source import CONTENT_SOURCE_ID, ContentSourceServices, FeedContentRuntime class FeedConfig(BaseModel): @@ -52,7 +52,7 @@ async def apply(ctx: Context, config: object) -> None: runtime = FeedContentRuntime( ctx.data_root, ctx.require(TIMERS), - ctx.require(CONTENT_SOURCE).bind("feed-subscriptions"), + ctx.require(CONTENT_SOURCE).bind(CONTENT_SOURCE_ID), ) def setup() -> object: diff --git a/tests/fixtures/legacy_feed_reservoir_rows.json b/tests/fixtures/legacy_feed_reservoir_rows.json new file mode 100644 index 0000000..6f6b555 --- /dev/null +++ b/tests/fixtures/legacy_feed_reservoir_rows.json @@ -0,0 +1,17 @@ +[ + {"item_id":"feed@github:subscriptions:event-01","kind":"content","source_id":"feed@github:subscriptions","original_source_id":"source-01","ack_source_id":"feed@github:subscriptions","source_event_id":"event-01","published_at":"2026-08-01T01:00:00+00:00","first_seen_at":"2026-08-01T01:01:00+00:00","preprocess_score":0.51,"payload_json":"{\"event_id\":\"event-01\",\"kind\":\"content\",\"preprocess_score\":0.51,\"preprocess_features\":{\"freshness\":0.91}}","embedding_json":null,"status":"unread","consumed_at":null}, + {"item_id":"feed@github:subscriptions:event-02","kind":"content","source_id":"feed@github:subscriptions","original_source_id":"source-02","ack_source_id":"feed@github:subscriptions","source_event_id":"event-02","published_at":"2026-08-01T02:00:00+00:00","first_seen_at":"2026-08-01T02:01:00+00:00","preprocess_score":0.52,"payload_json":"{\"event_id\":\"event-02\",\"kind\":\"content\",\"preprocess_score\":0.52,\"preprocess_features\":{\"freshness\":0.92}}","embedding_json":"[0.2,0.8]","status":"unread","consumed_at":null}, + {"item_id":"feed@github:subscriptions:event-03","kind":"content","source_id":"feed@github:subscriptions","original_source_id":"source-03","ack_source_id":"feed@github:subscriptions","source_event_id":"event-03","published_at":"2026-08-01T03:00:00+00:00","first_seen_at":"2026-08-01T03:01:00+00:00","preprocess_score":0.53,"payload_json":"{\"event_id\":\"event-03\",\"kind\":\"content\",\"preprocess_score\":0.53,\"preprocess_features\":{\"freshness\":0.93}}","embedding_json":null,"status":"unread","consumed_at":null}, + {"item_id":"feed@github:subscriptions:event-04","kind":"content","source_id":"feed@github:subscriptions","original_source_id":"source-04","ack_source_id":"feed@github:subscriptions","source_event_id":"event-04","published_at":"2026-08-01T04:00:00+00:00","first_seen_at":"2026-08-01T04:01:00+00:00","preprocess_score":0.54,"payload_json":"{\"event_id\":\"event-04\",\"kind\":\"content\",\"preprocess_score\":0.54,\"preprocess_features\":{\"freshness\":0.94}}","embedding_json":"[0.4,0.6]","status":"unread","consumed_at":null}, + {"item_id":"feed@github:subscriptions:event-05","kind":"content","source_id":"feed@github:subscriptions","original_source_id":"source-05","ack_source_id":"feed@github:subscriptions","source_event_id":"event-05","published_at":"2026-08-01T05:00:00+00:00","first_seen_at":"2026-08-01T05:01:00+00:00","preprocess_score":0.55,"payload_json":"{\"event_id\":\"event-05\",\"kind\":\"content\",\"preprocess_score\":0.55,\"preprocess_features\":{\"freshness\":0.95}}","embedding_json":null,"status":"unread","consumed_at":null}, + {"item_id":"feed@github:subscriptions:event-06","kind":"content","source_id":"feed@github:subscriptions","original_source_id":"source-06","ack_source_id":"feed@github:subscriptions","source_event_id":"event-06","published_at":"2026-08-01T06:00:00+00:00","first_seen_at":"2026-08-01T06:01:00+00:00","preprocess_score":0.56,"payload_json":"{\"event_id\":\"event-06\",\"kind\":\"content\",\"preprocess_score\":0.56,\"preprocess_features\":{\"freshness\":0.96}}","embedding_json":"[0.6,0.4]","status":"unread","consumed_at":null}, + {"item_id":"feed@github:subscriptions:event-07","kind":"content","source_id":"feed@github:subscriptions","original_source_id":"source-07","ack_source_id":"feed@github:subscriptions","source_event_id":"event-07","published_at":"2026-08-01T07:00:00+00:00","first_seen_at":"2026-08-01T07:01:00+00:00","preprocess_score":0.57,"payload_json":"{\"event_id\":\"event-07\",\"kind\":\"content\",\"preprocess_score\":0.57,\"preprocess_features\":{\"freshness\":0.97}}","embedding_json":null,"status":"unread","consumed_at":null}, + {"item_id":"feed@github:subscriptions:event-08","kind":"content","source_id":"feed@github:subscriptions","original_source_id":"source-08","ack_source_id":"feed@github:subscriptions","source_event_id":"event-08","published_at":"2026-08-01T08:00:00+00:00","first_seen_at":"2026-08-01T08:01:00+00:00","preprocess_score":0.58,"payload_json":"{\"event_id\":\"event-08\",\"kind\":\"content\",\"preprocess_score\":0.58,\"preprocess_features\":{\"freshness\":0.98}}","embedding_json":"[0.8,0.2]","status":"unread","consumed_at":null}, + {"item_id":"feed@github:subscriptions:event-09","kind":"content","source_id":"feed@github:subscriptions","original_source_id":"source-09","ack_source_id":"feed@github:subscriptions","source_event_id":"event-09","published_at":"2026-08-01T09:00:00+00:00","first_seen_at":"2026-08-01T09:01:00+00:00","preprocess_score":0.59,"payload_json":"{\"event_id\":\"event-09\",\"kind\":\"content\",\"preprocess_score\":0.59,\"preprocess_features\":{\"freshness\":0.99}}","embedding_json":null,"status":"unread","consumed_at":null}, + {"item_id":"feed@github:subscriptions:event-10","kind":"content","source_id":"feed@github:subscriptions","original_source_id":"source-10","ack_source_id":"feed@github:subscriptions","source_event_id":"event-10","published_at":"2026-08-01T10:00:00+00:00","first_seen_at":"2026-08-01T10:01:00+00:00","preprocess_score":0.60,"payload_json":"{\"event_id\":\"event-10\",\"kind\":\"content\",\"preprocess_score\":0.6,\"preprocess_features\":{\"freshness\":1.0}}","embedding_json":"[1.0,0.0]","status":"unread","consumed_at":null}, + {"item_id":"feed@github:subscriptions:event-11","kind":"content","source_id":"feed@github:subscriptions","original_source_id":"source-11","ack_source_id":"feed@github:subscriptions","source_event_id":"event-11","published_at":"2026-08-01T11:00:00+00:00","first_seen_at":"2026-08-01T11:01:00+00:00","preprocess_score":0.61,"payload_json":"{\"event_id\":\"event-11\",\"kind\":\"content\",\"preprocess_score\":0.61,\"preprocess_features\":{\"freshness\":0.89}}","embedding_json":null,"status":"unread","consumed_at":null}, + {"item_id":"feed@github:subscriptions:event-12","kind":"content","source_id":"feed@github:subscriptions","original_source_id":"source-12","ack_source_id":"feed@github:subscriptions","source_event_id":"event-12","published_at":"2026-08-01T12:00:00+00:00","first_seen_at":"2026-08-01T12:01:00+00:00","preprocess_score":0.62,"payload_json":"{\"event_id\":\"event-12\",\"kind\":\"content\",\"preprocess_score\":0.62,\"preprocess_features\":{\"freshness\":0.88}}","embedding_json":"[0.3,0.7]","status":"unread","consumed_at":null}, + {"item_id":"feed@github:subscriptions:event-13","kind":"content","source_id":"feed@github:subscriptions","original_source_id":"source-13","ack_source_id":"feed@github:subscriptions","source_event_id":"event-13","published_at":"2026-08-01T13:00:00+00:00","first_seen_at":"2026-08-01T13:01:00+00:00","preprocess_score":0.63,"payload_json":"{\"event_id\":\"event-13\",\"kind\":\"content\",\"preprocess_score\":0.63,\"preprocess_features\":{\"freshness\":0.87}}","embedding_json":null,"status":"unread","consumed_at":null}, + {"item_id":"feed@github:subscriptions:event-14","kind":"content","source_id":"feed@github:subscriptions","original_source_id":"source-14","ack_source_id":"feed@github:subscriptions","source_event_id":"event-14","published_at":"2026-08-01T14:00:00+00:00","first_seen_at":"2026-08-01T14:01:00+00:00","preprocess_score":0.64,"payload_json":"{\"event_id\":\"event-14\",\"kind\":\"content\",\"preprocess_score\":0.64,\"preprocess_features\":{\"freshness\":0.86}}","embedding_json":"[0.7,0.3]","status":"unread","consumed_at":null}, + {"item_id":"feed@github:subscriptions:event-15","kind":"content","source_id":"feed@github:subscriptions","original_source_id":"source-15","ack_source_id":"feed@github:subscriptions","source_event_id":"event-15","published_at":"2026-08-01T15:00:00+00:00","first_seen_at":"2026-08-01T15:01:00+00:00","preprocess_score":0.65,"payload_json":"{\"event_id\":\"event-15\",\"kind\":\"content\",\"preprocess_score\":0.65,\"preprocess_features\":{\"freshness\":0.85}}","embedding_json":null,"status":"unread","consumed_at":null} +] diff --git a/tests/test_legacy_handoff.py b/tests/test_legacy_handoff.py new file mode 100644 index 0000000..fd29595 --- /dev/null +++ b/tests/test_legacy_handoff.py @@ -0,0 +1,324 @@ +from __future__ import annotations + +import hashlib +import json +import sqlite3 +from collections.abc import Mapping, Sequence +from contextlib import closing +from datetime import UTC, datetime +from pathlib import Path +from typing import cast + +import pytest + +from agent.migrations.proactive_island import ( + HandoffBlocked, + HandoffStatus, + Inventory, + LegacyFact, + LegacyFactKind, + apply_handoff, +) +from plugins.content.store import ContentIdentityConflict, ContentStore + +from feed_runtime import backend +from legacy_handoff import FeedLegacyHandoffAdapter, LEGACY_SOURCE_ID + + +ROOT = Path(__file__).resolve().parent +TARGET_SOURCE = "feed-subscriptions" + + +class _BoundContent: + def __init__(self, store: ContentStore, source_id: str = TARGET_SOURCE) -> None: + self.store = store + self.source_id = source_id + + def submit( + self, batch_id: str, items: Sequence[Mapping[str, object]] + ) -> Mapping[str, object]: + return self.store.submit(self.source_id, batch_id, items) + + def read_submission(self, batch_id: str) -> Mapping[str, object] | None: + return self.store.read_submission(self.source_id, batch_id) + + def read_revision( + self, item_id: str, revision: str + ) -> Mapping[str, object] | None: + return self.store.read_revision(self.source_id, item_id, revision) + + def unsettled(self, limit: int = 100) -> tuple[Mapping[str, object], ...]: + return self.store.unsettled(self.source_id, limit) + + def ack(self, settlement_ref: str) -> Mapping[str, object]: + return self.store.ack(self.source_id, settlement_ref) + + +def _legacy_rows() -> list[dict[str, object]]: + value = json.loads( + (ROOT / "fixtures" / "legacy_feed_reservoir_rows.json").read_text() + ) + assert isinstance(value, list) and len(value) == 15 + return cast(list[dict[str, object]], value) + + +def _fact(row: Mapping[str, object]) -> LegacyFact: + opaque = json.dumps( + row, ensure_ascii=False, sort_keys=True, separators=(",", ":") + ).encode("utf-8") + item_id = cast(str, row["item_id"]) + return LegacyFact( + kind=LegacyFactKind.WAKE_SOURCE_ITEM, + locator=f"wake:reservoir_events:{item_id}", + source_digest=hashlib.sha256(opaque).hexdigest(), + source_identity="feed@github:subscriptions", + opaque=opaque, + ) + + +def _seed_provider(data_root: Path, rows: Sequence[Mapping[str, object]]) -> None: + config = backend.load_config(data_root) + connection = backend._connect(config) + try: + for index, row in enumerate(rows, start=1): + event_id = cast(str, row["source_event_id"]) + connection.execute( + """ + INSERT INTO items( + event_id, source_id, source_name, source_type, title, + content, url, author, published_at, first_seen_at, + last_seen_at, emitted_at, content_hash + ) VALUES(?, ?, ?, 'rss', ?, ?, ?, ?, ?, ?, ?, NULL, ?) + """, + ( + event_id, + f"source-{index:02d}", + f"Source {index:02d}", + f"Title {index:02d}", + f"Body {index:02d}", + f"https://example.com/{index:02d}", + None if index % 3 == 0 else f"Author {index:02d}", + row["published_at"], + row["first_seen_at"], + row["first_seen_at"], + f"revision-{index:02d}", + ), + ) + connection.commit() + finally: + connection.close() + + +def _fixture( + tmp_path: Path, +) -> tuple[Path, ContentStore, _BoundContent, FeedLegacyHandoffAdapter]: + data_root = tmp_path / "feed-data" + _seed_provider(data_root, _legacy_rows()) + store = ContentStore(tmp_path / "content.sqlite3") + store.initialize() + bound = _BoundContent(store) + return data_root, store, bound, FeedLegacyHandoffAdapter(data_root, bound) + + +def _tree_state(root: Path) -> tuple[tuple[str, int, int, str], ...]: + return tuple( + ( + str(path.relative_to(root)), + path.stat().st_size, + path.stat().st_mtime_ns, + hashlib.sha256(path.read_bytes()).hexdigest(), + ) + for path in sorted(root.rglob("*")) + if path.is_file() + ) + + +def test_real_shape_fifteen_rows_plan_apply_and_verify(tmp_path: Path) -> None: + data_root, store, _bound, adapter = _fixture(tmp_path) + facts = tuple(_fact(row) for row in _legacy_rows()) + before = _tree_state(tmp_path) + + plans = tuple(adapter.plan(fact) for fact in facts) + + assert _tree_state(tmp_path) == before + assert len({plan.target_identity for plan in plans}) == 15 + receipts = tuple( + adapter.apply(fact, plan) for fact, plan in zip(facts, plans, strict=True) + ) + after_apply = _tree_state(tmp_path) + assert all( + adapter.verify(fact, receipt) + for fact, receipt in zip(facts, receipts, strict=True) + ) + assert _tree_state(tmp_path) == after_apply + assert len({receipt.receipt_id for receipt in receipts}) == 15 + assert store.state_counts() == {"pending": 15} + with closing(sqlite3.connect(store.path)) as connection: + assert connection.execute( + "SELECT DISTINCT source_id FROM items" + ).fetchall() == [(TARGET_SOURCE,)] + assert not (data_root / "legacy_handoff.log").exists() + + +def test_adapter_accepts_only_legacy_feed_content_fact(tmp_path: Path) -> None: + _data_root, _store, _bound, adapter = _fixture(tmp_path) + fact = _fact(_legacy_rows()[0]) + + assert adapter.accepts(fact) is True + assert fact.source_identity == LEGACY_SOURCE_ID + assert adapter.accepts( + LegacyFact( + fact.kind, + fact.locator, + fact.source_digest, + "calendar@github:upcoming", + fact.opaque, + ) + ) is False + + +def test_missing_provider_item_blocks_without_writes(tmp_path: Path) -> None: + data_root, store, _bound, adapter = _fixture(tmp_path) + config = backend.load_config(data_root) + connection = backend._connect(config) + connection.execute("DELETE FROM items WHERE event_id='event-15'") + connection.commit() + connection.close() + before = _tree_state(tmp_path) + + with pytest.raises(HandoffBlocked, match="feed_provider_item_missing:event-15"): + adapter.plan(_fact(_legacy_rows()[-1])) + + assert _tree_state(tmp_path) == before + assert store.state_counts() == {} + + +def test_target_before_marker_replay_returns_same_receipt(tmp_path: Path) -> None: + _data_root, store, _bound, adapter = _fixture(tmp_path) + fact = _fact(_legacy_rows()[0]) + plan = adapter.plan(fact) + + first = adapter.apply(fact, plan) + repeated = adapter.apply(fact, plan) + + assert repeated == first + assert adapter.verify(fact, repeated) is True + assert store.state_counts() == {"pending": 1} + + +def test_core_replays_after_target_receipt_before_lineage_marker( + tmp_path: Path, +) -> None: + _data_root, store, _bound, adapter = _fixture(tmp_path) + fact = _fact(_legacy_rows()[0]) + inventory = Inventory((fact,), ()) + workspace = tmp_path / "workspace" + + def crash_after_target(_fact: LegacyFact, _receipt: object) -> None: + raise RuntimeError("crash before central marker") + + with pytest.raises(RuntimeError, match="crash before central marker"): + apply_handoff( + workspace, + inventory, + (adapter,), + after_target=crash_after_target, + ) + assert store.state_counts() == {"pending": 1} + + recovered = apply_handoff(workspace, inventory, (adapter,)) + + assert recovered.status is HandoffStatus.APPLIED + assert recovered.items[0].state == "applied" + assert store.state_counts() == {"pending": 1} + + +def test_revision_change_after_target_is_a_batch_conflict(tmp_path: Path) -> None: + data_root, store, _bound, adapter = _fixture(tmp_path) + fact = _fact(_legacy_rows()[0]) + original = adapter.plan(fact) + _ = adapter.apply(fact, original) + config = backend.load_config(data_root) + connection = backend._connect(config) + connection.execute( + "UPDATE items SET content_hash='revision-changed' WHERE event_id='event-01'" + ) + connection.commit() + connection.close() + changed = adapter.plan(fact) + + with pytest.raises(ContentIdentityConflict, match="batch identity conflict"): + adapter.apply(fact, changed) + + assert store.state_counts() == {"pending": 1} + + +def test_plan_revision_change_before_apply_fails_before_submit(tmp_path: Path) -> None: + data_root, store, _bound, adapter = _fixture(tmp_path) + fact = _fact(_legacy_rows()[0]) + plan = adapter.plan(fact) + config = backend.load_config(data_root) + connection = backend._connect(config) + connection.execute( + "UPDATE items SET content_hash='revision-new' WHERE event_id='event-01'" + ) + connection.commit() + connection.close() + + with pytest.raises(RuntimeError, match="target identity drift"): + adapter.apply(fact, plan) + + assert store.state_counts() == {} + + +def test_already_acked_target_replays_and_source_ack_is_bound_once( + tmp_path: Path, +) -> None: + data_root, store, bound, adapter = _fixture(tmp_path) + now = datetime(2026, 8, 23, 10, tzinfo=UTC) + fact = _fact(_legacy_rows()[0]) + plan = adapter.plan(fact) + receipt = adapter.apply(fact, plan) + snapshot = store.snapshot(now) + candidate = cast(tuple[dict[str, object], ...], snapshot["items"])[0] + selected = store.select( + cast(Mapping[str, object], candidate["ref"]), + cast(int, snapshot["snapshot_seq"]), + {"session_id": "wake:fixture", "turn_id": "turn:feed-legacy"}, + now, + ) + token = cast(str, selected["selection_token"]) + _ = store.transition(token, "ready_for_delivery") + _ = store.transition(token, "delivered", settlement_ref="delivery:feed-legacy") + + assert _BoundContent(store, "another-source").unsettled() == () + assert len(bound.unsettled()) == 1 + assert backend.settle_content_item( + "event-01", "revision-01", data_root=data_root + )["disposition"] == "acknowledged" + assert bound.ack("delivery:feed-legacy") == { + "settled": True, + "duplicate": False, + } + assert bound.ack("delivery:feed-legacy") == { + "settled": True, + "duplicate": True, + } + assert adapter.apply(fact, plan) == receipt + assert adapter.verify(fact, receipt) is True + assert store.state_counts() == {"settled": 1} + + +def test_provider_plan_requires_checkpoint_and_creates_no_files( + tmp_path: Path, +) -> None: + data_root, store, _bound, adapter = _fixture(tmp_path) + wal = data_root / "feed_mcp.sqlite3-wal" + wal.write_bytes(b"not-checkpointed") + before = _tree_state(tmp_path) + + with pytest.raises(HandoffBlocked, match="feed_provider_checkpoint_required"): + adapter.plan(_fact(_legacy_rows()[0])) + + assert _tree_state(tmp_path) == before + assert store.state_counts() == {} From 1d2e7b34817c749853b03a0a9806bb1a39b56ff0 Mon Sep 17 00:00:00 2001 From: Codex Date: Sun, 23 Aug 2026 23:29:14 +0800 Subject: [PATCH 2/4] ci(feed): pin proactive handoff core --- .github/workflows/plugin-api-v3.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/plugin-api-v3.yml b/.github/workflows/plugin-api-v3.yml index 9b7412f..e11ad19 100644 --- a/.github/workflows/plugin-api-v3.yml +++ b/.github/workflows/plugin-api-v3.yml @@ -36,7 +36,7 @@ jobs: - uses: actions/checkout@v4 with: repository: kachofugetsu09/akashic-agent - ref: 9da3a988a2bf62b0f550bd4f6bb98c4eeb1f56f5 + ref: bd5db8c2f9f857c7b5a6e44abfbaf14a2f5485ee path: .akashic-core - uses: actions/setup-python@v5 with: From 018aa81938311ebd1059ebbd92d761c419776745 Mon Sep 17 00:00:00 2001 From: Codex Date: Sun, 23 Aug 2026 23:39:45 +0800 Subject: [PATCH 3/4] fix(feed): hand off pending wake acknowledgements --- .github/workflows/plugin-api-v3.yml | 4 +- feed_runtime/backend.py | 159 ++++++++++++++++++++++++++++ legacy_handoff.py | 103 +++++++++++++++--- pyrightconfig.json | 1 + tests/test_legacy_handoff.py | 70 ++++++++++++ 5 files changed, 321 insertions(+), 16 deletions(-) diff --git a/.github/workflows/plugin-api-v3.yml b/.github/workflows/plugin-api-v3.yml index e11ad19..3bd42a8 100644 --- a/.github/workflows/plugin-api-v3.yml +++ b/.github/workflows/plugin-api-v3.yml @@ -62,8 +62,8 @@ jobs: env: AKASHIC_AGENT_ROOT: .akashic-core PYTHONPATH: .akashic-core:mcp/.venv/lib/python3.13/site-packages - run: mcp/.venv/bin/pyright plugin.py content_source.py feed_runtime mcp/run_mcp.py mcp/src mcp/scripts scripts + run: mcp/.venv/bin/pyright plugin.py content_source.py legacy_handoff.py feed_runtime mcp/run_mcp.py mcp/src mcp/scripts scripts - name: Compile Python sources - run: python -m compileall -q plugin.py content_source.py feed_runtime mcp/run_mcp.py mcp/src mcp/scripts scripts tests + run: python -m compileall -q plugin.py content_source.py legacy_handoff.py feed_runtime mcp/run_mcp.py mcp/src mcp/scripts scripts tests - name: Check diff formatting run: git diff --check diff --git a/feed_runtime/backend.py b/feed_runtime/backend.py index 9f006e2..470f83b 100644 --- a/feed_runtime/backend.py +++ b/feed_runtime/backend.py @@ -1954,6 +1954,165 @@ def settle_content_item( conn.close() +def settle_legacy_ack( + event_id: str, + revision: str, + action: str, + source_digest: str, + *, + data_root: Path, +) -> dict[str, str]: + """Commit one legacy Wake ACK and retain its target-owned receipt.""" + + cfg = load_config(data_root) + conn = _connect(cfg) + now = _now() + receipt_id = f"feed-legacy-ack:{source_digest}" + try: + # 1. Reuse a completed handoff without extending the provider ACK. + _ensure_legacy_ack_receipts(conn) + existing = conn.execute( + "SELECT * FROM legacy_ack_handoff_receipts WHERE receipt_id = ?", + (receipt_id,), + ).fetchone() + if existing is not None: + identity = tuple( + str(existing[field]) + for field in ("source_digest", "event_id", "revision", "action") + ) + if identity != (source_digest, event_id, revision, action): + raise RuntimeError("Feed legacy ACK receipt identity conflict") + return _legacy_ack_receipt(existing) + + # 2. Commit one exact provider ACK and its durable receipt atomically. + acked_at, expires_at = _commit_legacy_provider_ack( + conn, cfg, event_id, revision, now + ) + _insert_legacy_ack_receipt( + conn, + receipt_id, + source_digest, + event_id, + revision, + action, + acked_at, + expires_at, + now, + ) + conn.commit() + row = conn.execute( + "SELECT * FROM legacy_ack_handoff_receipts WHERE receipt_id = ?", + (receipt_id,), + ).fetchone() + if row is None: + raise RuntimeError("Feed legacy ACK receipt commit missing") + return _legacy_ack_receipt(row) + finally: + conn.close() + + +def _ensure_legacy_ack_receipts(conn: sqlite3.Connection) -> None: + conn.execute( + """ + CREATE TABLE IF NOT EXISTS legacy_ack_handoff_receipts( + receipt_id TEXT PRIMARY KEY, + source_digest TEXT NOT NULL UNIQUE, + event_id TEXT NOT NULL, + revision TEXT NOT NULL, + action TEXT NOT NULL, + acked_at TEXT NOT NULL, + expires_at TEXT NOT NULL, + committed_at TEXT NOT NULL + ) + """ + ) + + +def _commit_legacy_provider_ack( + conn: sqlite3.Connection, + cfg: FeedMcpConfig, + event_id: str, + revision: str, + now: datetime, +) -> tuple[str, str]: + """Preserve a live provider ACK or establish one new retention window.""" + + current = conn.execute( + "SELECT content_hash FROM items WHERE event_id = ?", (event_id,) + ).fetchone() + if current is None: + raise RuntimeError(f"Feed legacy ACK provider item missing: {event_id}") + if str(current["content_hash"]) != revision: + raise RuntimeError(f"Feed legacy ACK revision changed: {event_id}") + acknowledgement = conn.execute( + "SELECT acked_at, expires_at FROM acked_items WHERE event_id = ?", + (event_id,), + ).fetchone() + if acknowledgement is not None and datetime.fromisoformat( + str(acknowledgement["expires_at"]) + ) > now: + return str(acknowledgement["acked_at"]), str(acknowledgement["expires_at"]) + acked_at = now.isoformat() + expires_at = (now + timedelta(hours=cfg.item_retention_hours)).isoformat() + conn.execute( + """ + INSERT INTO acked_items(event_id, acked_at, expires_at) + VALUES (?, ?, ?) + ON CONFLICT(event_id) DO UPDATE SET + acked_at=excluded.acked_at, + expires_at=excluded.expires_at + """, + (event_id, acked_at, expires_at), + ) + return acked_at, expires_at + + +def _insert_legacy_ack_receipt( + conn: sqlite3.Connection, + receipt_id: str, + source_digest: str, + event_id: str, + revision: str, + action: str, + acked_at: str, + expires_at: str, + now: datetime, +) -> None: + conn.execute( + """ + INSERT INTO legacy_ack_handoff_receipts( + receipt_id, source_digest, event_id, revision, action, + acked_at, expires_at, committed_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + receipt_id, + source_digest, + event_id, + revision, + action, + acked_at, + expires_at, + now.isoformat(), + ), + ) + + +def _legacy_ack_receipt(row: sqlite3.Row) -> dict[str, str]: + return { + field: str(row[field]) + for field in ( + "receipt_id", + "source_digest", + "event_id", + "revision", + "action", + "acked_at", + "expires_at", + ) + } + + def content_source_deadline(*, data_root: Path, now: datetime) -> datetime: """Return the durable next source deadline, defaulting to immediate work.""" diff --git a/legacy_handoff.py b/legacy_handoff.py index 61e066c..6a3a9a1 100644 --- a/legacy_handoff.py +++ b/legacy_handoff.py @@ -45,28 +45,39 @@ def __init__(self, feed_data_root: Path, content: BoundContentSource) -> None: def accepts(self, fact: LegacyFact) -> bool: return ( - fact.kind is LegacyFactKind.WAKE_SOURCE_ITEM + fact.kind in {LegacyFactKind.WAKE_SOURCE_ITEM, LegacyFactKind.WAKE_ACK} and fact.source_identity == LEGACY_SOURCE_ID ) def plan(self, fact: LegacyFact) -> AdapterPlan: """Resolve one provider-owned revision without mounting or writing Content.""" - row = _legacy_row(fact) + row = _fact_row(fact) provider = self._provider_item(_text(row, "source_event_id")) - return AdapterPlan(_target_identity(provider)) + return AdapterPlan(_target_identity(fact.kind, provider)) def apply(self, fact: LegacyFact, plan: AdapterPlan) -> TargetReceipt: """Submit the exact planned target and return its normalized durable receipt.""" # 1. Re-read the owner row and reject a revision change after planning. - row = _legacy_row(fact) + row = _fact_row(fact) provider = self._provider_item(_text(row, "source_event_id")) - target_identity = _target_identity(provider) + target_identity = _target_identity(fact.kind, provider) if plan.target_identity != target_identity: raise RuntimeError("Feed handoff target identity drift after plan") - # 2. A fact-stable batch makes target-before-marker replay idempotent. + # 2. Each fact kind commits through its target owner's durable primitive. + if fact.kind is LegacyFactKind.WAKE_ACK: + acknowledgement = backend.settle_legacy_ack( + _text(provider, "event_id"), + _text(provider, "content_hash"), + _text(row, "action"), + fact.source_digest, + data_root=self._provider_db.parent, + ) + return _ack_receipt(fact, target_identity, acknowledgement) + + # 3. A fact-stable batch makes target-before-marker replay idempotent. batch_id = _batch_id(fact) item = _content_item(row, provider) content_receipt = self._content.submit(batch_id, (item,)) @@ -87,8 +98,14 @@ def verify(self, fact: LegacyFact, receipt: TargetReceipt) -> bool: return False if receipt.target_identity != plan.target_identity: return False - row = _legacy_row(fact) + row = _fact_row(fact) provider = self._provider_item(_text(row, "source_event_id")) + if fact.kind is LegacyFactKind.WAKE_ACK: + acknowledgement = self._provider_ack(fact.source_digest) + if acknowledgement is None: + return False + expected = _ack_receipt(fact, plan.target_identity, acknowledgement) + return receipt == expected item = _content_item(row, provider) batch_id = _batch_id(fact) submission = self._content.read_submission(batch_id) @@ -104,6 +121,28 @@ def verify(self, fact: LegacyFact, receipt: TargetReceipt) -> bool: and _revision_matches(revision, item) ) + def _provider_ack(self, source_digest: str) -> dict[str, object] | None: + """Read one retained legacy ACK receipt without opening a writer.""" + + wal = self._provider_db.with_name(self._provider_db.name + "-wal") + if wal.is_file() and wal.stat().st_size > 0: + raise HandoffBlocked("feed_provider_checkpoint_required") + uri = self._provider_db.resolve().as_uri() + "?mode=ro&immutable=1" + with closing(sqlite3.connect(uri, uri=True)) as connection: + connection.row_factory = sqlite3.Row + _ = connection.execute("PRAGMA query_only = ON") + table = connection.execute( + "SELECT 1 FROM sqlite_master " + "WHERE type='table' AND name='legacy_ack_handoff_receipts'" + ).fetchone() + if table is None: + return None + result = connection.execute( + "SELECT * FROM legacy_ack_handoff_receipts WHERE source_digest = ?", + (source_digest,), + ).fetchone() + return None if result is None else {key: result[key] for key in result.keys()} + def _provider_item(self, event_id: str) -> dict[str, object]: """Read one exact Feed row through a query-only SQLite connection.""" @@ -130,8 +169,8 @@ def _provider_item(self, event_id: str) -> dict[str, object]: return {key: result[key] for key in result.keys()} -def _legacy_row(fact: LegacyFact) -> dict[str, object]: - if fact.kind is not LegacyFactKind.WAKE_SOURCE_ITEM: +def _fact_row(fact: LegacyFact) -> dict[str, object]: + if fact.kind not in {LegacyFactKind.WAKE_SOURCE_ITEM, LegacyFactKind.WAKE_ACK}: raise TypeError("Feed handoff received another legacy fact kind") if fact.source_identity != LEGACY_SOURCE_ID: raise TypeError("Feed handoff received another legacy source owner") @@ -144,9 +183,15 @@ def _legacy_row(fact: LegacyFact) -> dict[str, object]: event_id = _text(row, "source_event_id") if not fact.locator.endswith(f":{_text(row, 'item_id')}"): raise RuntimeError("Feed legacy locator does not match item_id") - payload = _payload(row) - if payload.get("event_id") != event_id or payload.get("kind") != "content": - raise RuntimeError("Feed legacy payload identity mismatch") + if fact.kind is LegacyFactKind.WAKE_SOURCE_ITEM: + payload = _payload(row) + if payload.get("event_id") != event_id or payload.get("kind") != "content": + raise RuntimeError("Feed legacy payload identity mismatch") + elif _text(row, "source_id") != LEGACY_SOURCE_ID or _text(row, "action") not in { + "consume", + "expire", + }: + raise RuntimeError("Feed legacy ACK identity mismatch") return row @@ -182,13 +227,43 @@ def _content_item( } -def _target_identity(provider: Mapping[str, object]) -> str: +def _target_identity(kind: LegacyFactKind, provider: Mapping[str, object]) -> str: + prefix = "content" if kind is LegacyFactKind.WAKE_SOURCE_ITEM else "feed-ack" return ( - f"content:{CONTENT_SOURCE_ID}:{_text(provider, 'event_id')}:" + f"{prefix}:{CONTENT_SOURCE_ID}:{_text(provider, 'event_id')}:" f"{_text(provider, 'content_hash')}" ) +def _ack_receipt( + fact: LegacyFact, + target_identity: str, + acknowledgement: Mapping[str, object], +) -> TargetReceipt: + normalized = { + "legacy_locator": fact.locator, + "legacy_source_digest": fact.source_digest, + "target_identity": target_identity, + "acknowledgement": { + key: acknowledgement[key] + for key in ( + "receipt_id", + "source_digest", + "event_id", + "revision", + "action", + "acked_at", + "expires_at", + ) + }, + } + return TargetReceipt( + receipt_id=_text(acknowledgement, "receipt_id"), + receipt_digest=receipt_digest(normalized), + target_identity=target_identity, + ) + + def _batch_id(fact: LegacyFact) -> str: encoded = f"{fact.locator}\x00{fact.source_digest}".encode("utf-8") return f"feed-legacy:{hashlib.sha256(encoded).hexdigest()}" diff --git a/pyrightconfig.json b/pyrightconfig.json index 7c1ff38..e676354 100644 --- a/pyrightconfig.json +++ b/pyrightconfig.json @@ -2,6 +2,7 @@ "include": [ "plugin.py", "content_source.py", + "legacy_handoff.py", "feed_runtime", "mcp/run_mcp.py", "mcp/src", diff --git a/tests/test_legacy_handoff.py b/tests/test_legacy_handoff.py index fd29595..b63babd 100644 --- a/tests/test_legacy_handoff.py +++ b/tests/test_legacy_handoff.py @@ -76,6 +76,31 @@ def _fact(row: Mapping[str, object]) -> LegacyFact: ) +def _ack_fact(row: Mapping[str, object], action: str = "consume") -> LegacyFact: + event_id = cast(str, row["source_event_id"]) + item_id = cast(str, row["item_id"]) + acknowledgement = { + "source_id": LEGACY_SOURCE_ID, + "source_event_id": event_id, + "item_id": item_id, + "action": action, + "queued_at": "2026-08-23T10:00:00+00:00", + } + opaque = json.dumps( + acknowledgement, ensure_ascii=False, sort_keys=True, separators=(",", ":") + ).encode("utf-8") + return LegacyFact( + kind=LegacyFactKind.WAKE_ACK, + locator=( + "wake:pending_acknowledgements:" + f"{LEGACY_SOURCE_ID}:{event_id}:{item_id}" + ), + source_digest=hashlib.sha256(opaque).hexdigest(), + source_identity=LEGACY_SOURCE_ID, + opaque=opaque, + ) + + def _seed_provider(data_root: Path, rows: Sequence[Mapping[str, object]]) -> None: config = backend.load_config(data_root) connection = backend._connect(config) @@ -309,6 +334,51 @@ def test_already_acked_target_replays_and_source_ack_is_bound_once( assert store.state_counts() == {"settled": 1} +@pytest.mark.parametrize("action", ["consume", "expire"]) +def test_core_hands_off_pending_ack_once_after_target_replay( + tmp_path: Path, + action: str, +) -> None: + data_root, store, _bound, adapter = _fixture(tmp_path) + fact = _ack_fact(_legacy_rows()[0], action) + inventory = Inventory((fact,), ()) + workspace = tmp_path / "workspace" + + def crash_after_target(_fact: LegacyFact, _receipt: object) -> None: + raise RuntimeError("crash before central ACK marker") + + with pytest.raises(RuntimeError, match="crash before central ACK marker"): + apply_handoff( + workspace, + inventory, + (adapter,), + after_target=crash_after_target, + ) + + config = backend.load_config(data_root) + with closing(backend._connect(config)) as connection: + assert connection.execute("SELECT count(*) FROM acked_items").fetchone()[0] == 1 + assert connection.execute( + "SELECT count(*) FROM legacy_ack_handoff_receipts" + ).fetchone()[0] == 1 + connection.execute("DELETE FROM acked_items") + connection.commit() + + recovered = apply_handoff(workspace, inventory, (adapter,)) + + assert recovered.status is HandoffStatus.APPLIED + assert recovered.items[0].state == "applied" + assert store.state_counts() == {} + with closing(backend._connect(config)) as connection: + assert connection.execute("SELECT count(*) FROM acked_items").fetchone()[0] == 0 + assert [ + tuple(row) + for row in connection.execute( + "SELECT event_id, revision, action FROM legacy_ack_handoff_receipts" + ).fetchall() + ] == [("event-01", "revision-01", action)] + + def test_provider_plan_requires_checkpoint_and_creates_no_files( tmp_path: Path, ) -> None: From eec005f80088228291647116b08133ecfa4818ef Mon Sep 17 00:00:00 2001 From: Codex Date: Sun, 23 Aug 2026 23:42:48 +0800 Subject: [PATCH 4/4] fix(feed): preserve configured handoff root --- legacy_handoff.py | 3 ++- tests/test_legacy_handoff.py | 17 +++++++++++++++++ 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/legacy_handoff.py b/legacy_handoff.py index 6a3a9a1..a267902 100644 --- a/legacy_handoff.py +++ b/legacy_handoff.py @@ -40,6 +40,7 @@ class FeedLegacyHandoffAdapter: """Move exact legacy Feed reservoir facts into the existing Content source.""" def __init__(self, feed_data_root: Path, content: BoundContentSource) -> None: + self._data_root = feed_data_root self._provider_db = backend.provider_database_path(feed_data_root) self._content = content @@ -73,7 +74,7 @@ def apply(self, fact: LegacyFact, plan: AdapterPlan) -> TargetReceipt: _text(provider, "content_hash"), _text(row, "action"), fact.source_digest, - data_root=self._provider_db.parent, + data_root=self._data_root, ) return _ack_receipt(fact, target_identity, acknowledgement) diff --git a/tests/test_legacy_handoff.py b/tests/test_legacy_handoff.py index b63babd..f698b94 100644 --- a/tests/test_legacy_handoff.py +++ b/tests/test_legacy_handoff.py @@ -392,3 +392,20 @@ def test_provider_plan_requires_checkpoint_and_creates_no_files( assert _tree_state(tmp_path) == before assert store.state_counts() == {} + + +def test_pending_ack_uses_original_root_for_nested_provider_path( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + config = backend._config_values() + config["db_path"] = "state/feed.sqlite3" + monkeypatch.setattr(backend, "_config_values", lambda: config) + data_root, _store, _bound, adapter = _fixture(tmp_path) + fact = _ack_fact(_legacy_rows()[0]) + + receipt = adapter.apply(fact, adapter.plan(fact)) + + assert adapter.verify(fact, receipt) is True + assert (data_root / "state" / "feed.sqlite3").is_file() + assert not (data_root / "state" / "state" / "feed.sqlite3").exists()