Phase 3C.1.5: async store refactor (complete)#8
Merged
Conversation
…e.py Adds AsyncExtractionPipeline class using ChainStoreProtocol. Sync ExtractionPipeline is left in place for sync callers in subscriptions/batch/cli.py/entity_ops/exporter - those migrate in later tasks. test_pipeline.py now uses the new async pipeline + async_chain_stores fixture. Also folds the deferred Phase 1 cleanup: AsyncChainStore cache methods (extraction_cache + llm_link_cache) now filter and populate the user_id column now that migration v4 has merged. The SQLite NULL-safe filter pattern (user_id IS ? OR user_id = ?) is used so None and real-UUID callers both match their own rows. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Adds AsyncLinkerEngine class using ChainStoreProtocol. Replaces per-partner fetch loop with single batch fetch_findings_by_ids call (spec G6). Wraps relation upsert in store.transaction() for atomicity. Also folds deferred Phase 1 cleanup: set_run_status now writes to the linker_run.status_text column (migration v4) instead of the temporary in-memory _run_status dict. start_linker_run initializes new rows with status_text='pending'. LinkerRun model gains a status field populated from status_text by _row_to_linker_run. test_set_run_status_updates_in_memory_status was updated to assert on the persisted status via fetch_linker_runs rather than peeking at the removed _run_status attribute, and renamed accordingly. Sync LinkerEngine is left in place for consumers not yet migrated (subscriptions, batch context, cli.py rebuild, test_linker_batch, test_llm_pass, test_subscriptions). Removed in Task 30. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
llm_link_pass_async now uses ChainStoreProtocol methods (fetch_relations_in_scope, apply_link_classification, LLM cache get/put). Wraps per-edge classification + cache write in store.transaction() for atomicity. Uses link_classification_cache_key from the consolidated _cache_keys module. Sync llm_link_pass left in place until Task 30 for callers not yet migrated. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Tasks 22a/b/c landed (AsyncExtractionPipeline, AsyncLinkerEngine, async llm_link_pass). Suite at 612 passed, 1 skipped. Next session resumes at Task 22d (AsyncChainBatchContext). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Parallel to sync ChainBatchContext. Stage 1 fetches all deferred findings in a single query; stage 2 runs extraction in parallel via asyncio.gather + Semaphore(4); stage 3 links sequentially. Per-finding transactions so partial progress is visible and the batch is crash-resilient. Sync ChainBatchContext remains for callers not yet migrated (subscriptions drain worker migration in Task 22e, CLI in 22f). Deleted in Task 30. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
start_drain_worker returns a DrainWorker handle that CLI lifecycle awaits on shutdown. Sync event bus handler queues finding ids via loop.call_soon_threadsafe so it is safe from any thread context. Short-circuits when _in_batch_context is True so batch-mode callers can own end-to-end processing. Sync subscribe_chain_handlers path remains in place for test fixtures not yet migrated (5 existing sync tests continue to pass). Deleted in Task 30 alongside the other sync removals. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
_get_stores_async opens an AsyncChainStore against the default CLI database. rebuild now runs the async rebuild pipeline (AsyncExtractionPipeline + AsyncLinkerEngine), enumerating findings via fetch_findings_for_engagement / fetch_findings_by_ids protocol methods (no more raw SQL), and closes the store in a finally block. Typer 0.24 does not natively support async def commands, so the body is wrapped in asyncio.run inside a sync command. The async implementation lives in _rebuild_async. For the "all engagements" path we enumerate via EngagementStore.list_all() (there is no global list_findings helper) and fan out scoped protocol queries. Other CLI commands (status, entities, path, export, query) remain sync until their dependencies (entity_ops, exporter, query engine, graph cache) are converted in later Phase 2+ tasks. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Session 2 landed Tasks 22d (AsyncChainBatchContext), 22e (drain worker), 22f (async rebuild command), and 22g verification. Phase 2 is now complete: all pipeline/linker/batch/subscriptions/rebuild sync callers have parallel async implementations. Suite at 614 passed, 1 skipped. Next session resumes at Task 23 (entity_ops). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…t_findings Three followups from the Phase 2 session 2 handoff: 1. CLI async commands now use an @_async_command decorator that wraps an async body in asyncio.run while preserving Typer's option introspection via functools.wraps. Typer 0.24.1 silently discards bare "async def" commands (the coroutine is created but never awaited), so the decorator is the portable fix. _rebuild_async is gone - the rebuild body is inline under the decorator. 2. DrainWorker.wait_idle() pumps pending call_soon_threadsafe dispatches via asyncio.sleep(0) and then awaits queue.join(). This replaces the flaky sleep(0.01) hack in the two drain worker tests with a deterministic one-line API. DrainWorker.stop() now routes through wait_idle as well. 3. EngagementStore.list_findings() returns every non-deleted finding across all engagements in one call. cli.py rebuild's "all engagements" path no longer has to enumerate engagements and fan out per-engagement fetches. Suite: 614 passed, 1 skipped (unchanged). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
merge_entities and split_entity now use ChainStoreProtocol methods (fetch_mentions_with_engagement, rewrite_mentions_entity_id, rewrite_mentions_by_ids, delete_entity, upsert_entity, recompute_mention_counts) instead of raw SQL. Each operation is wrapped in store.batch_transaction() for atomicity. test_entity_ops.py converted to async using async_chain_stores fixture and AsyncExtractionPipeline for seeding. All 6 tests pass. MergeResult.affected_findings is now returned as an empty list: the protocol's fetch_mentions_with_engagement returns only (mention_id, engagement_id) tuples without finding_id, and no "distinct findings for entity" helper exists on the protocol. No current callers or tests depend on that field; a future task can extend the protocol if needed. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
export_chain and import_chain now use ChainStoreProtocol. Export streams rows via store.export_dump_stream for bounded memory and decodes BLOB columns (reasons_json, confirmed_at_reasons_json) into embedded JSON so the dump is self-describing. Import reconstructs Entity / EntityMention / FindingRelation domain models from the dump and calls upsert_entities_bulk, add_mentions_bulk, upsert_relations_bulk inside store.batch_transaction() for atomicity on partial failure. Adds fetch_all_finding_ids to ChainStoreProtocol and AsyncChainStore for the exporter's "all engagements" path. Protocol method count: 41 -> 42. test_store_protocol_shape updated accordingly. CLI export command converted to async via the @_async_command decorator from the Phase 2 gotcha fixes. test_cli_export_runs kept sync: CliRunner invokes the decorated wrapper, which runs asyncio.run() internally — there's no outer event loop in a sync test so that path just works. test_exporter converted to async via module-level pytestmark. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Both commands use @_async_command decorator and _get_stores_async helper. Raw SQL replaced with list_entities, fetch_relations_in_scope, and fetch_linker_runs protocol methods. Existing sync test_cli_status_runs / test_cli_entities_runs / test_cli_entities_filter_by_type tests continue to pass unchanged (per the CliRunner + asyncio.run pattern documented in Task 24: CLI tests that invoke commands through CliRunner stay sync because asyncio.run() cannot be called from a pytest-asyncio outer loop). path and query commands remain sync pending Phase 4 GraphCache / ChainQueryEngine conversions. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Session 3 landed Phase 2 gotcha fixes (f7134e1), Task 23 async entity_ops (609bfd7), Task 24 async exporter + CLI export (ef127a1), and Task 25 async CLI status/entities (8a66666). Suite at 614 passed, 1 skipped throughout. Next session resumes at Task 26 (async GraphCache with per-key asyncio.Lock). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…CLI)
Bundles Tasks 26-29 from the revised Phase 3C.1.5 plan into one commit
because GraphCache's consumers (ChainQueryEngine, presets, cli.py
path/query) share a tight dependency chain - splitting would produce
red intermediate commits.
- GraphCache.get_master_graph is now async with a per-key asyncio.Lock
(spec G4). Concurrent callers for the same (user_id, generation,
include_candidates, include_rejected) key collapse to a single
_build_master_graph invocation; waiters re-check the cache under the
lock before returning. Uses stream_relations_in_scope +
fetch_findings_by_ids protocol methods; no more raw SQL against the
sync ChainStore.
- Two new protocol methods on ChainStoreProtocol /
AsyncChainStore drive the query stack:
- fetch_finding_ids_for_entity for entity-kind endpoint resolution
- fetch_entity_mentions_for_engagement for external_to_internal and
mitre_coverage presets.
- ChainQueryEngine.k_shortest_paths and query.endpoints.resolve_endpoint
are now async; Yen's algorithm, virtual super-source/sink wiring, and
predicate filtering stay sync (all in-memory).
- All five built-in presets (lateral_movement, priv_esc_chains,
external_to_internal, crown_jewel, mitre_coverage) are async; they
use the new protocol methods instead of ad-hoc SQL.
- cli.py path and query commands use the @_async_command decorator +
_get_stores_async helper and await the async query stack.
- Test conversions:
- test_graph_cache.py: 10 -> 11 tests (+1 concurrent-build test that
patches _build_master_graph with a counting wrapper, gathers 10
racing get_master_graph calls, and asserts build_count == 1 and
all results are the same instance).
- test_query_engine.py, test_neighborhood.py, test_presets.py:
pytestmark = pytest.mark.asyncio + async_chain_stores fixture +
AsyncExtractionPipeline / AsyncLinkerEngine seeding.
- test_endpoints.py: sync tests use asyncio.run around resolve_endpoint;
the store-backed test moves to async_chain_stores.
- test_pipeline_integration.py: keeps sync ExtractionPipeline /
LinkerEngine seeding (valid until Task 30) and opens a separate
AsyncChainStore against the same DB file for the query/preset
sanity checks.
- test_store_protocol_shape.py: EXPECTED_METHODS grows from 42 to 44.
- test_cli_commands.py path/query tests unchanged (sync CliRunner +
asyncio.run pattern per Tasks 22f/24/25).
Baseline 614 -> 615 passed, 1 skipped. Narration, sync
ExtractionPipeline/LinkerEngine/ChainBatchContext/store_extensions, and
cli._get_stores stay for Task 30, which will delete the parallel sync
classes in one sweep.
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
narrate_path now uses store.get_llm_link_cache / put_llm_link_cache protocol methods and narration_cache_key from the consolidated _cache_keys module (deferred Phase 1 cleanup 3). test_narration.py converted to async using async_chain_stores fixture and AsyncExtractionPipeline for any needed seeding. Unblocks Task 30's store_extensions.py deletion. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Final retirement pass for Phase 2-4. Every downstream consumer is now async via ChainStoreProtocol, so the parallel sync classes can be deleted: - ExtractionPipeline (was AsyncExtractionPipeline) - LinkerEngine (was AsyncLinkerEngine) - ChainBatchContext (was AsyncChainBatchContext) - llm_link_pass (was llm_link_pass_async) Also deleted: - store_extensions.py (sync ChainStore / SyncChainStore alias) - sync subscribe_chain_handlers + factory-injection path from subscriptions.py (kept DrainWorker, start_drain_worker, set_batch_context, reset_subscriptions) - sync _get_stores helper from cli.py (renamed _get_stores_async -> _get_stores) - sync chain_store + engagement_store_and_chain fixtures from conftest.py (renamed async_chain_stores -> engagement_store_and_chain) test_subscriptions.py dropped its 5 sync factory-injection tests (idempotent, no_factories_is_noop, inline_handler_extracts_and_links, batch_context_suppresses_inline, disabled_config_skips_subscription). The 2 drain worker tests already cover the event-bus -> extraction pipeline end-to-end. Net test count: 615 -> 610. test_store.py and test_pipeline_integration.py were ported from the deleted sync ChainStore + execute_all/execute_one helpers onto AsyncChainStore protocol methods. test_cli_commands.py seeds its fixture via asyncio.run around the async pipeline/engine. All Phase 2-4 test files renamed to use the canonical engagement_store_and_chain fixture. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Bundles Phase 5 Tasks 31-37 into a single commit because the PostgresChainStore implementation is only testable once Task 37's conformance parameter is wired up, and partial implementations leave the suite in a non-functional state. - packages/cli/src/opentools/chain/stores/postgres_async.py: new PostgresChainStore class implementing all 44 ChainStoreProtocol methods using SQLAlchemy async against web SQLModel chain tables. Handles both sqlite+aiosqlite and postgresql dialects for upsert ON CONFLICT via a dialect-aware _insert_for() helper. The finding_extraction_state and finding_parser_output methods return no-op values since the web backend has not migrated those tables yet (a future phase will). - packages/web/backend/alembic/versions/004_chain_jsonb_unlogged_userids.py: new Alembic migration that creates chain_extraction_cache and chain_llm_link_cache with user_id columns, converts JSON TEXT columns to JSONB on Postgres (for chain_finding_relation reasons and chain_linker_run rule_stats), and marks cache tables UNLOGGED on Postgres (spec O17). On SQLite (the conformance path) the JSONB and UNLOGGED steps are skipped via dialect inspection. - packages/web/backend/app/models.py: ChainExtractionCache and ChainLlmLinkCache SQLModel table classes added to mirror the CLI schema so PostgresChainStore can query them via the ORM. - packages/cli/src/opentools/chain/stores/__init__.py: lazy export of PostgresChainStore so the CLI doesn't pay the web SQLModel import cost unless a caller actually asks for it. - packages/cli/tests/chain/test_store_protocol_conformance.py: postgres_async parameter activated. The fixture constructs a PostgresChainStore over sqlite+aiosqlite via SQLAlchemy, creates the web schema via SQLModel.metadata.create_all, and seeds a User row so foreign keys hold. Every conformance test now runs twice. The CLI-only test_upsert_and_get_extraction_hash test skips on the postgres path because the extraction-state table has not yet been added to the web schema. Test count: 610 -> 623 passed (+12 new postgres conformance tests, +1 skipped because finding_extraction_state is CLI-only for now). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Bundles Phase 5 Tasks 38-41 into a single commit: - chain_store_factory.py (new): chain_store_from_session and chain_store_from_factory helpers construct PostgresChainStore around a request-scoped AsyncSession or an async_sessionmaker. - chain_service.py: every method delegates to PostgresChainStore via the factory. create_linker_run_stub renamed to create_linker_run_pending and uses store.start_linker_run (with back-compat alias). k_shortest_paths_stub replaced with a real ChainQueryEngine + GraphCache call (CLI parity). - chain_rebuild_worker.py (new): run_rebuild_shared uses the shared ExtractionPipeline + LinkerEngine + PostgresChainStore against a background-task-scoped session from async_session_factory. Replaces the duplicated custom extractor in chain_rebuild.py. - routes/chain.py: rebuild endpoint launches run_rebuild_shared and uses create_linker_run_pending. Old chain_rebuild import removed. - chain_rebuild.py: deleted. No more duplicated pipeline code in the web backend. - test_chain_rebuild.py renamed to test_web_rebuild.py with assertions adapted to shared-pipeline output (6 linker rules, not just shared_strong_entity). Failure path patches LinkerEngine.make_context since the worker now wraps per-finding calls in try/except, and the outer handler records the failure. - test_pipeline_integration.py parameterized over both sqlite_async and postgres_async backends. sqlite_async seeds via EngagementStore (CLI path, user_id=None). postgres_async seeds via SQLModel ORM against a sqlite+aiosqlite SQLAlchemy session, yields a real user_id. Raw-SQL helpers replaced with protocol-based queries (list_entities + fetch_relations_in_scope). mitre_coverage kept as a sqlite-only check because it hardcodes user_id=None which PostgresChainStore rejects. Test count: 625 passed, 2 skipped (up from 623 baseline, net +2 from the 2 new parameterized integration-test variants). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
All remaining tasks (28b, 30, 31-37, 38-41, 42) landed this session. The Phase 3C.1.5 async store refactor is complete: 44-method ChainStoreProtocol with AsyncChainStore + PostgresChainStore, a single shared ExtractionPipeline/LinkerEngine backing both CLI and web, zero remaining sync chain code, and 625 tests passing (0 regressions across 17 commits since main). Ready to merge. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Fix 1 - MergeResult.affected_findings regression: entity_ops.merge_entities was returning affected_findings=[] with a NOTE that no protocol helper exposed "distinct findings for an entity". The protocol already has fetch_finding_ids_for_entity (added in Task 26 for the query engine's endpoint resolver) -- wire merge_entities to call it BEFORE rewriting mentions and populate the field. Added test_merge_affected_findings_spans_multiple_findings and tightened test_merge_two_host_entities to assert on the new data. Fix 3 - chain_rebuild_worker direct SQL escape hatch: run_rebuild_shared's failure path dropped to a raw SQLAlchemy UPDATE to mark the linker run failed because finish_linker_run expects full success counters. Added mark_run_failed(run_id, *, error, user_id) to ChainStoreProtocol with matching aiosqlite and SQLAlchemy-async implementations; the worker now routes its failure finalize through the protocol. EXPECTED_METHODS 44 -> 45. Fix 4 - finding_extraction_state / finding_parser_output web tables: test_upsert_and_get_extraction_hash was skipped on the postgres_async parameter because PostgresChainStore's upsert_extraction_state / get_extraction_hash / get_parser_output were no-op stubs and the web schema lacked the backing tables. Added Alembic migration 005 and SQLModel classes ChainFindingExtractionState + ChainFindingParserOutput, both user-scoped via nullable FK per spec G37. Replaced the stubs with real SQLAlchemy implementations that mirror AsyncChainStore's semantics (dialect-aware upsert via _insert_for, JSON blob via _coerce_json_bytes). Unskipped the test and added two new conformance tests (test_mark_run_failed_sets_status_and_error, test_fetch_finding_ids_for_entity_distinct) that run against both backends. Baseline 625 passed / 2 skipped -> 631 passed / 1 skipped. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…Fix 2)
Every ChainService method (list_entities, get_entity,
relations_for_finding, get_linker_run, create_linker_run_pending)
now delegates to PostgresChainStore and converts CLI domain
objects to response dicts via a new chain_dto.py converter module.
Zero remaining SQLModel ORM escape hatches in the chain service
layer; the public API JSON shape is preserved because the DTOs
export the same field names the old SQLModel row construction
consumed (including a status_text alias for ChainLinkerRun so
/rebuild and /runs/{id} keep reading the field they always did).
Closes the deferred follow-up noted in the session 4 handoff:
"Web chain_service read path still uses SQLModel ORM directly
(awaiting route-level DTO converter)".
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
- Conformance fixture now reads WEB_TEST_DB_URL at test time. When set (CI only), the postgres_async parameter runs against a real Postgres server at that URL instead of sqlite+aiosqlite. Per-test isolation is enforced by a fresh random user_id and a teardown block that deletes every chain row scoped to that user. Local pytest runs are unchanged (the env var is unset by default). - .github/workflows/postgres-tests.yml: new workflow that spins up postgres:16-alpine as a service container, runs alembic upgrade head, sets WEB_TEST_DB_URL, and runs the chain conformance suite plus web integration tests plus the full pytest packages/ baseline against real Postgres. Triggered on PRs touching packages/cli/** or packages/web/backend/** and on push to main. Closes the deferred follow-up noted in the session 4 handoff: "Real Postgres validation gated on WEB_TEST_DB_URL (CI should add a real-Postgres run for JSONB/UNLOGGED coverage)". The workflow gives the PostgresChainStore SQLAlchemy implementation real dialect coverage for JSONB columns, UNLOGGED table behavior, and asyncpg driver quirks that sqlite+aiosqlite cannot catch. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Migrations 002-005 referenced sqlmodel.sql.sqltypes.GUID for user_id
foreign key columns. That attribute was removed from newer sqlmodel
versions (confirmed missing in 0.0.38 locally), so alembic upgrade head
crashes with AttributeError on any fresh database. The existing
SQLite test suite never hit this because tests use
SQLModel.metadata.create_all() and bypass Alembic entirely; the new
postgres-tests workflow exposed the regression on its first real CI
run against Postgres.
Replaces every sa.Column("user_id", sqlmodel.sql.sqltypes.GUID(), ...)
with sa.Column("user_id", sa.Uuid(), ...). sa.Uuid is the native
SQLAlchemy 2.0 UUID type and works on both SQLite (stored as CHAR(32))
and PostgreSQL (stored as native UUID). Drops the now-unused
'import sqlmodel' from 002-005; 001_initial.py keeps it for AutoString.
No behavior change on SQLite (tests already pass against the in-memory
schema via SQLModel.metadata.create_all, which uses the SQLModel column
types directly). Fixes the alembic upgrade head step in the
postgres-tests workflow on PR #8.
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
The first real-Postgres CI run on PR #8 surfaced a pre-existing bug in the web backend's SQLModel table definitions. Most datetime fields are typed 'datetime' without an explicit sa_column override, so SQLModel infers 'DateTime()' (timezone=False) for them. SQLAlchemy's bind converter then strips tz info from any tz-aware Python datetime before handing it to asyncpg. Since every Alembic migration declares the underlying Postgres column as 'sa.DateTime(timezone=True)' (TIMESTAMPTZ), asyncpg receives a naive datetime for a TZ-aware column and raises 'DataError: can't subtract offset-naive and offset-aware datetimes' inside its timestamp encoder. Annotating each of the ~22 datetime fields across models.py with 'sa_column_kwargs={type_: DateTime(timezone=True)}' would be a larger and more intrusive change. Instead this commit installs a 'before_cursor_execute' event listener on the production AsyncEngine (and on the conformance fixture's engine) that promotes any naive datetime bind param to UTC-aware right before the DBAPI sees it. Idempotent on already-aware values and a no-op on SQLite (which stores both variants as ISO strings). Works for single-row and executemany bind parameters alike. Unblocks the postgres-tests CI workflow introduced in 6aa2f9d. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Previous fix in 0f33237 installed the before_cursor_execute listener but didn't account for two asyncpg dialect realities: 1. The asyncpg dialect uses positional tuple parameters by default, not dicts. The old listener had a dict-only branch and never mutated tuples, so every Insert against the real engine passed through unchanged. 2. Tuples are immutable, so in-place mutation isn't an option even if the branch had existed. The listener must return the modified (statement, parameters) pair and be registered with retval=True. This commit rewrites stamp_naive_datetimes_utc to handle dicts, tuples, lists, and executemany-style list-of-rows uniformly, and registers the listener with retval=True on both the production engine and the conformance fixture engine. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…ield Replaces the before_cursor_execute listener shim from 0f33237/811756e which did not actually fix the asyncpg tz-naive datetime issue -- the SQLAlchemy asyncpg dialect applies its own parameter conversion AFTER before_cursor_execute fires, so the listener's fix-up was silently undone. This commit does the model-level fix instead: a new TZAwareDateTime TypeDecorator is declared at the top of models.py, and every datetime field across the 14 SQLModel tables (22 fields total) is updated to use sa_type=TZAwareDateTime via a shared _TZ_KW kwargs dict. The TypeDecorator: - declares its impl as DateTime(timezone=True), so DDL generates TIMESTAMPTZ on Postgres regardless of the Python-side annotation - stamps tz-naive values with UTC in process_bind_param before any dialect processor sees them - stamps tz-naive values with UTC in process_result_value so reads always hand tz-aware datetimes back to Python Idempotent on already-tz-aware values and a no-op on SQLite, which stores both variants as ISO strings regardless. The removed before_cursor_execute shim is gone from database.py and from the conformance test fixture. Local suite still at 631 passed, 1 skipped. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
asyncpg rejects pre-serialized orjson strings against a JSONB column with 'DatatypeMismatchError: column is of type jsonb but expression is of type character varying'. The SQLModel tables declare reasons_json / confirmed_at_reasons_json / rule_stats_json as Column(Text), and PostgresChainStore writes orjson-serialized strings, so the JSONB conversion in migration 004 was a schema-vs-code mismatch waiting to bite on real Postgres. Reverting the JSONB ALTER COLUMN leaves the columns TEXT on Postgres, matching SQLite behavior. No code path uses JSONB-specific operators on these columns, so the conversion was a nice-to-have optimization rather than a requirement (spec O17 is about UNLOGGED cache tables, which this migration still sets). Unblocks the last 4 failing postgres_async conformance tests: - test_start_and_finish_linker_run - test_current_linker_generation_monotone - test_mark_run_failed_sets_status_and_error - test_fetch_finding_ids_for_entity_distinct Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…een tests Two tail-end postgres_async conformance failures, both only visible against real Postgres: 1. _jsonb_dumps returned orjson.dumps() bytes, which asyncpg rejects when binding to a VARCHAR/Text column with 'expected str, got bytes'. SQLite is lax about bytes vs str and silently stored the bytes as its own str-ish wrapper, hiding the bug. Now decode to UTF-8 before returning. 2. Conformance fixture teardown only purged chain_* tables, not the hardcoded Engagement and Finding rows that _seed_finding_row inserts with primary-key ids 'eng_conf' / 'fnd_conf' / 'fnd_conf_2'. On SQLite each test starts from a fresh tmp_path DB so the collision never happens; against real Postgres every test after the first hit 'duplicate key value violates unique constraint engagement_pkey'. The teardown loop now also deletes Finding + Engagement rows scoped to the test's user_id, in FK-safe order (child tables first). Expected to clear the last 2 postgres_async conformance failures: test_start_and_finish_linker_run and test_fetch_finding_ids_for_entity_distinct. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
The class defines 'async def list(self, ...)' which shadows the builtin list at class-body scope. The later 'async def search(...) -> list[Finding]' annotation tries to subscript that shadowed method, which on Python 3.12 (CI) raises 'TypeError: function object is not subscriptable' at class definition time. Python 3.14 (local) tolerates it because annotations are lazily evaluated by default, which is why the local suite never caught this. 'from __future__ import annotations' makes every annotation in the module a string evaluated lazily by typing.get_type_hints(), which sidesteps the class-body lookup entirely. One-line fix, zero runtime behavior change. Surfaced by the new postgres-tests.yml CI workflow running on Python 3.12.13. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Same root cause as e765ac5: several service classes define an 'async def list(self, ...)' method which shadows the builtin list at class-body scope. Later methods use 'list[SomeType]' as a return annotation, which Python 3.12 evaluates eagerly at class-def time and resolves against the shadowed attribute, raising 'function object is not subscriptable'. Python 3.14 tolerates this via lazy annotation evaluation. Adding 'from __future__ import annotations' to every service module forces all annotations in the module to be strings and lazily evaluated, sidestepping the class-body lookup. Applied defensively to every service file even if only two currently trip the bug: - engagement_service (has list()) - ioc_service (has list(), tripped CI) - correlation_service (defensive) - recipe_service (defensive) finding_service already got the fix in e765ac5. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…win32 test_check_all_returns_report hardcoded 'win32' as the expected report.platform, which passes on the Windows dev machine and fails on Linux CI. The runner.check_all() method just returns sys.platform, so asserting against sys.platform is the correct cross-platform check. Surfaced by the new postgres-tests CI workflow running on ubuntu-latest. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Async-first refactor of the chain data layer. Every chain code path now uses
ChainStoreProtocolwith two implementations:AsyncChainStore(aiosqlite, CLI) andPostgresChainStore(SQLAlchemy async, ~1060 lines, all 46 methods — web backend). Single sharedExtractionPipeline+LinkerEngine+ChainBatchContext+llm_link_passbacking both CLI and web, replacing the web backend's previously duplicated custom extractor.Test count
613 (main baseline) → 631 (+18 net), 0 regressions across 22 commits. The one remaining skip is the unrelated
test_llm_providers.pyLLM smoke test disabled by design.Key outcomes
ChainStoreProtocol+ two fully compliant backendspostgres_asyncparam usessqlite+aiosqliteby default; CI runs it against real Postgres viaWEB_TEST_DB_URL)DrainWorker.wait_idle()helper with deterministicasyncio.sleep(0)semantics forcall_soon_threadsafepumpingstatus_text+chain_finding_extraction_state+chain_finding_parser_outputtablesPostgresChainStore; newchain_dto.pyconverter layer preserves the public JSON wire shapeDeferred follow-ups
All 5 deferred follow-ups from the post-Phase-5 handoff are closed in this branch:
Test plan
Session handoff docs
🤖 Generated with Claude Code