diff --git a/benchmarks/adapters/conversation.py b/benchmarks/adapters/conversation.py index b9a697d..84eeb95 100644 --- a/benchmarks/adapters/conversation.py +++ b/benchmarks/adapters/conversation.py @@ -463,14 +463,27 @@ async def _process_one(item_index: int) -> Tuple[bool, Optional[str]]: self._lme_session_to_uri[session_id] = uris[0] return (True, None) + except asyncio.CancelledError: + # Per-item cancellation must not cascade to siblings + # (REVIEW REL-04). Surface as a structured error. + return (False, f"item={item_index}: cancelled") except Exception as exc: return (False, f"item={item_index}: {exc}") + # ``return_exceptions=True`` keeps a single bad item from + # aborting the rest of the benchmark run — the inner try/except + # turns recoverable failures into structured tuples, and any + # BaseException that does escape appears as an exception object + # we surface alongside the structured results. results = await asyncio.gather( *[_process_one(idx) for idx in selected_indices], - return_exceptions=False, + return_exceptions=True, ) - for ok, err in results: + for result in results: + if isinstance(result, BaseException): + errors.append(f"item=?: {result!r}") + continue + ok, err = result if ok: ingested += 1 elif err: diff --git a/benchmarks/adapters/locomo.py b/benchmarks/adapters/locomo.py index 0ff9b8e..652e7a5 100644 --- a/benchmarks/adapters/locomo.py +++ b/benchmarks/adapters/locomo.py @@ -576,14 +576,29 @@ async def _process_one(conv: Dict[str, Any]) -> Tuple[bool, Optional[str]]: if uri in new_records ] return (True, None) + except asyncio.CancelledError: + # Per-conversation cancellation must not cascade to + # siblings (REVIEW REL-04). Surface as an error so + # the gather summary reports it, but do not propagate. + return (False, f"conv={conv_id}: cancelled") except Exception as exc: return (False, f"conv={conv_id}: {exc}") + # ``return_exceptions=True`` ensures one cancelled or crashed + # conversation does not abort sibling work — the error string + # is captured per-conversation by the inner try/except, and + # any exception that escapes (typically only BaseException + # subclasses we cannot swallow) appears in ``results`` as an + # exception object that we surface alongside the others. results = await asyncio.gather( *[_process_one(conv) for conv in conversations], - return_exceptions=False, + return_exceptions=True, ) - for ok, err in results: + for result in results: + if isinstance(result, BaseException): + errors.append(f"conv=?: {result!r}") + continue + ok, err = result if ok: ingested += 1 elif err: diff --git a/src/opencortex/context/manager.py b/src/opencortex/context/manager.py index dcb273a..635952c 100644 --- a/src/opencortex/context/manager.py +++ b/src/opencortex/context/manager.py @@ -77,6 +77,33 @@ ] # (collection, tenant_id, user_id, session_id, turn_id) +class RecompositionError(Exception): + """Raised by ``_run_full_session_recomposition`` with raise_on_error=True + when work fails partway through. + + Carries the directory URIs that were already created before the + failure so the caller can register them with its run-scoped cleanup + tracker (REVIEW REL-02). Without this, raise_on_error=True callers + saw the inner ``contextlib.suppress`` best-effort cleanup as a + silent black box: any URI that failed to delete inside the + suppressed block became an orphan with no signal to the outer + layer. + + Callers should drain ``created_uris`` into their tracker, then + ``raise exc.original from exc`` (or just ``raise``) so the outer + handler still sees the underlying failure. + """ + + def __init__( + self, + original: BaseException, + created_uris: List[str], + ) -> None: + super().__init__(str(original)) + self.original = original + self.created_uris = list(created_uris) + + class SourceConflictError(Exception): """Same session_id ingested twice with a different transcript. @@ -1220,6 +1247,103 @@ async def _persist_rendered_conversation_source( ) return source_uri + async def _mark_source_run_complete(self, source_uri: str) -> None: + """Set ``meta.run_complete=True`` on a benchmark source record. + + REVIEW F5 / ADV-007: written only after a benchmark ingest run + finishes successfully. The idempotent-hit path treats a + hash-match WITHOUT this marker as a torn prior run (compensate + partially failed, leaving stale leaves) and re-ingests after + purging, instead of silently returning the partial set. The + marker write goes through the lower-level storage update so we + avoid the heavy ``MemoryOrchestrator.update`` re-derive path — + this is meta-only. + """ + if not source_uri: + return + try: + records = await self._orchestrator._storage.filter( + self._orchestrator._get_collection(), + {"op": "must", "field": "uri", "conds": [source_uri]}, + limit=1, + ) + if not records: + return + record = records[0] + record_id = str(record.get("id", "") or "") + if not record_id: + return + existing_meta = dict(record.get("meta") or {}) + existing_meta["run_complete"] = True + await self._orchestrator._storage.update( + self._orchestrator._get_collection(), + record_id, + {"meta": existing_meta}, + ) + except Exception as exc: # pragma: no cover - defensive log path + logger.warning( + "benchmark_ingest: failed to mark source %s run_complete: %s", + source_uri, + exc, + exc_info=True, + ) + + async def _purge_torn_benchmark_run( + self, + *, + session_id: str, + tenant_id: str, + user_id: str, + source_uri: str, + merged_records: List[Dict[str, Any]], + ) -> None: + """Drop stale records left by a prior failed benchmark ingest run. + + Used when the idempotent-hit path detects a hash-match without + the ``run_complete`` marker (REVIEW F5). Removes: + + - merged leaves under the source (the records we already loaded) + - directory records under the source (re-loaded here) + - the deterministic session_summary record if one exists + + The source record itself stays — it is keyed deterministically + on (tenant, user, session_id) and the next ingest will rewrite + the transcript_hash and ``run_complete`` marker fresh. + """ + merged_uris = [ + str(rec.get("uri", "") or "").strip() + for rec in merged_records + if rec.get("uri") + ] + try: + directory_records = await self._load_session_directory_records( + session_id=session_id, + source_uri=source_uri, + ) + except Exception: # pragma: no cover - defensive + directory_records = [] + directory_uris = [ + str(rec.get("uri", "") or "").strip() + for rec in directory_records + if rec.get("uri") + ] + summary_uri = self._session_summary_uri(tenant_id, user_id, session_id) + + all_uris = [u for u in (merged_uris + directory_uris + [summary_uri]) if u] + if not all_uris: + return + logger.info( + "benchmark_ingest: purging torn prior run sid=%s source=%s " + "merged=%d directories=%d summary=%s", + session_id, + source_uri, + len(merged_uris), + len(directory_uris), + "present" if summary_uri else "absent", + ) + with contextlib.suppress(Exception): + await self._delete_immediate_families(all_uris) + @classmethod def _benchmark_segment_meta( cls, @@ -1493,30 +1617,84 @@ async def benchmark_ingest_conversation( source_uri=source_uri, ) if existing_records: - logger.info( - "benchmark_ingest_conversation: idempotent hit " - "sid=%s source_uri=%s records=%d", - session_id, - source_uri, - len(existing_records), + # Verify the prior run actually completed (REVIEW F5 / + # ADV-007). The transcript hash matches but if the + # ``run_complete`` marker is absent, the prior ingest + # crashed before it could mark itself done — the + # records on disk are a partial / torn set, not a + # genuine idempotent hit. Treat as cold ingest after + # purging the stale records. + source_record = await self._orchestrator._get_record_by_uri( + source_uri ) - hydrated = await self._hydrate_record_contents(existing_records) - return { - "status": "ok", - "session_id": session_id, - "source_uri": source_uri, - "summary_uri": None, - "records": [ - self._export_memory_record( - record, - hydrated_content=hydrated.get( - str(record.get("uri", "") or ""), - "", - ), + source_meta = ( + dict(source_record.get("meta") or {}) if source_record else {} + ) + run_complete = bool(source_meta.get("run_complete")) + + if not run_complete: + logger.warning( + "benchmark_ingest_conversation: torn prior run " + "detected sid=%s source=%s — purging stale " + "records and re-ingesting", + session_id, + source_uri, + ) + await self._purge_torn_benchmark_run( + session_id=session_id, + tenant_id=tenant_id, + user_id=user_id, + source_uri=source_uri, + merged_records=existing_records, + ) + # fall through to the cold ingest path below + else: + # Look up the prior run's session_summary if one + # exists under the deterministic URI (REVIEW F4 / + # api-contract-005). Previously the idempotent path + # always returned summary_uri=None even when the + # first run had persisted a summary, leaving an + # asymmetric contract that could surprise downstream + # consumers. + existing_summary_uri = self._session_summary_uri( + tenant_id, user_id, session_id + ) + existing_summary = await ( + self._orchestrator._get_record_by_uri( + existing_summary_uri ) - for record in existing_records - ], - } + ) + summary_uri_for_response = ( + existing_summary_uri if existing_summary else None + ) + + logger.info( + "benchmark_ingest_conversation: idempotent hit " + "sid=%s source_uri=%s records=%d summary=%s", + session_id, + source_uri, + len(existing_records), + "present" if existing_summary else "absent", + ) + hydrated = await self._hydrate_record_contents( + existing_records + ) + return { + "status": "ok", + "session_id": session_id, + "source_uri": source_uri, + "summary_uri": summary_uri_for_response, + "records": [ + self._export_memory_record( + record, + hydrated_content=hydrated.get( + str(record.get("uri", "") or ""), + "", + ), + ) + for record in existing_records + ], + } cleanup = _BenchmarkRunCleanup(source_uri=source_uri) # In-memory map URI -> raw conversation text for U10 content @@ -1628,14 +1806,27 @@ async def _bounded_complete( raise if cleanup.merged_uris: - directory_uris = await self._run_full_session_recomposition( - session_id=session_id, - tenant_id=tenant_id, - user_id=user_id, - source_uri=source_uri, - raise_on_error=True, - return_created_uris=True, - ) + try: + directory_uris = ( + await self._run_full_session_recomposition( + session_id=session_id, + tenant_id=tenant_id, + user_id=user_id, + source_uri=source_uri, + raise_on_error=True, + return_created_uris=True, + ) + ) + except RecompositionError as exc: + # Drain partial URIs into the tracker so the + # outer except handler's compensate() removes + # them via the same code path that handles + # successful directory writes (REVIEW REL-02). + # Re-raise the original exception so the outer + # handler still sees the underlying failure + # type (CancelledError, RuntimeError, ...). + cleanup.directory_uris.extend(exc.created_uris) + raise exc.original from exc if directory_uris: cleanup.directory_uris.extend(directory_uris) if include_session_summary: @@ -1674,6 +1865,13 @@ async def _bounded_complete( if fallback_targets else {} ) + # Mark the source as run_complete BEFORE returning so an + # immediate retry sees the marker and short-circuits via + # the idempotent path (REVIEW F5 / ADV-007). Failure to + # mark is logged but does not fail the response — the + # next replay would just trigger a redundant purge + + # re-ingest, which is correct, just slower. + await self._mark_source_run_complete(source_uri) return { "status": "ok", "session_id": session_id, @@ -1805,6 +2003,10 @@ async def _benchmark_ingest_direct_evidence( if record: records.append(record) + # Mark source as run_complete so the next replay short-circuits + # via the idempotent path (REVIEW F5 / ADV-007). Symmetric with + # the merged_recompose return path above. + await self._mark_source_run_complete(source_uri or "") return { "status": "ok", "session_id": session_id, @@ -2940,11 +3142,21 @@ async def _derive_one( exc, exc_info=True, ) + if raise_on_error: + # Hand back the partial URIs to the caller via + # RecompositionError so its run-scoped cleanup tracker + # can drive compensation (REVIEW REL-02). The previous + # inline ``contextlib.suppress`` cleanup was a silent + # black box on the raise_on_error=True path: any URI + # that failed to delete inside the suppressed block + # became an orphan with no signal upstream. + raise RecompositionError(exc, created_directory_uris) from exc + # raise_on_error=False (production lifecycle) keeps the + # legacy best-effort inline cleanup. Production callers do + # not maintain a tracker and rely on this fallback. if created_directory_uris: with contextlib.suppress(Exception): await self._delete_immediate_families(created_directory_uris) - if raise_on_error: - raise if return_created_uris: # Failure path with reporting requested: hand back what # we created before the failure so the caller's tracker diff --git a/src/opencortex/orchestrator.py b/src/opencortex/orchestrator.py index 9d9ec03..d59127e 100644 --- a/src/opencortex/orchestrator.py +++ b/src/opencortex/orchestrator.py @@ -5243,17 +5243,35 @@ async def benchmark_conversation_ingest( segments: List[List[Dict[str, Any]]], include_session_summary: bool = True, ingest_shape: str = "merged_recompose", + enforce_admin: bool = True, ) -> Dict[str, Any]: """Public facade for benchmark-only offline conversation ingest. Delegates to :meth:`ContextManager.benchmark_ingest_conversation`. - Admin-gated at the HTTP layer; this method itself does not enforce - identity policy because non-HTTP callers (tests, in-process - benchmark drivers) are trusted by construction. + + Defense-in-depth: by default this method requires the request + role contextvar to be ``admin`` (REVIEW api-contract-007). The + HTTP layer's ``_require_admin()`` already gates the only + production caller, but the facade is now public on the + orchestrator, so a future internal caller cannot accidentally + bypass the policy by skipping the HTTP route. Direct in-process + callers (benchmark CLI runs that pre-seed the role contextvar, + unit tests, or maintenance scripts) can pass + ``enforce_admin=False`` to opt out explicitly — the kwarg is + deliberately verbose so the bypass shows up in code review. """ self._ensure_init() if not self._context_manager: raise RuntimeError("ContextManager not initialized") + if enforce_admin: + from opencortex.http.request_context import is_admin + + if not is_admin(): + raise PermissionError( + "benchmark_conversation_ingest requires admin role " + "(set request role contextvar to 'admin' or pass " + "enforce_admin=False for trusted in-process callers)" + ) return await self._context_manager.benchmark_ingest_conversation( session_id=session_id, tenant_id=tenant_id, diff --git a/tests/test_benchmark_ingest_lifecycle.py b/tests/test_benchmark_ingest_lifecycle.py index 5452739..32776a0 100644 --- a/tests/test_benchmark_ingest_lifecycle.py +++ b/tests/test_benchmark_ingest_lifecycle.py @@ -140,6 +140,93 @@ async def check(): self._run(check()) + def test_torn_prior_run_is_not_treated_as_idempotent(self): + """Hash-match without run_complete marker triggers purge + re-ingest. + + Simulates a torn prior run by writing the source first (so the + hash matches on replay) but stripping the ``run_complete`` meta + marker. The next ingest with the same payload must NOT take the + idempotent short-circuit; it must purge stale records and run + a fresh ingest. + """ + + async def check(): + import opencortex.http.server as http_server + + async with _test_app_context() as client: + role_token = set_request_role("admin") + try: + # Successful first ingest writes records and marks + # the source as run_complete. + payload = _payload("bench_torn_01") + r1 = await client.post(_ADMIN_INGEST_URL, json=payload) + self.assertEqual(r1.status_code, 200) + first = r1.json() + + # Strip the run_complete marker to simulate a torn + # prior run (compensate failed midway). + orch = http_server._orchestrator + storage = orch._storage + coll = orch._get_collection() + src_records = await storage.filter( + coll, + { + "op": "must", + "field": "uri", + "conds": [first["source_uri"]], + }, + limit=1, + ) + self.assertTrue(src_records) + src = src_records[0] + src_meta = dict(src.get("meta") or {}) + src_meta.pop("run_complete", None) + await storage.update( + coll, str(src["id"]), {"meta": src_meta} + ) + + # Replay with the same payload — should NOT short-circuit. + # Track _purge_torn_benchmark_run to confirm it ran. + cm = orch._context_manager + purges: list = [] + original_purge = cm._purge_torn_benchmark_run + + async def _track_purge(**kwargs): + purges.append(kwargs.get("source_uri")) + await original_purge(**kwargs) + + cm._purge_torn_benchmark_run = _track_purge + try: + r2 = await client.post(_ADMIN_INGEST_URL, json=payload) + finally: + cm._purge_torn_benchmark_run = original_purge + + self.assertEqual(r2.status_code, 200) + self.assertEqual( + purges, + [first["source_uri"]], + "torn-replay must trigger _purge_torn_benchmark_run", + ) + # Re-ingest should produce a fresh record set with + # the marker re-applied. + src_records2 = await storage.filter( + coll, + { + "op": "must", + "field": "uri", + "conds": [first["source_uri"]], + }, + limit=1, + ) + self.assertTrue( + src_records2[0].get("meta", {}).get("run_complete"), + "run_complete must be re-set after successful re-ingest", + ) + finally: + reset_request_role(role_token) + + self._run(check()) + # ----------------------------------------------------------------- # AR4 — Cleanup tracker for failure / cancellation paths # -----------------------------------------------------------------