feat: two-pass critique extractor — bench-validated, parked

packages/python-sdk-memwal/memwal/client.py

@@ -614,6 +614,7 @@ async def analyze(
         text: str,
         namespace: Optional[str] = None,
         occurred_at: Optional[Union[str, datetime]] = None,
+        extract_with_critique: bool = False,
     ) -> AnalyzeResult:
         """Analyze conversation text and return as soon as facts are accepted.
 
@@ -645,6 +646,15 @@ async def analyze(
                 (no ``now()`` fallback). The resolved date lives only
                 inside the encrypted fact text + embedding; there is no
                 server-readable metadata column for it (Architecture A).
+            extract_with_critique: Opt into the two-pass extraction
+                critique. When ``True``, the server runs a
+                second LLM pass that critiques and corrects the
+                first-pass extracted facts before any fact is embedded
+                or stored. The critic sees the same ``occurred_at``
+                anchor and related-memories context the first pass
+                saw. **Doubles the per-analyze LLM call count** —
+                default ``False`` preserves the single-pass write
+                path. Wire field is omitted when ``False``.
 
         Returns:
             :class:`AnalyzeResult` with extracted ``facts`` + per-fact
@@ -657,6 +667,11 @@ async def analyze(
         wire_occurred_at = _occurred_at_to_wire(occurred_at)
         if wire_occurred_at is not None:
             body["occurred_at"] = wire_occurred_at
+        # Only include the critique flag when explicitly True; matches
+        # the server's `#[serde(default)]` shape so the wire payload is
+        # byte-identical for existing callers.
+        if extract_with_critique:
+            body["extract_with_critique"] = True
         data = await self._signed_request(
             "POST",
             "/api/analyze",
@@ -690,6 +705,7 @@ async def analyze_and_wait(
         namespace: Optional[str] = None,
         opts: Optional[RememberBulkOptions] = None,
         occurred_at: Optional[Union[str, datetime]] = None,
+        extract_with_critique: bool = False,
     ) -> AnalyzeWaitResult:
         """Analyze + wait for every extracted fact to finish persisting.
 
@@ -698,11 +714,16 @@ async def analyze_and_wait(
         result combines the analyze fact list with the bulk-style settled
         per-job results.
 
-        ``occurred_at`` carries the same temporal-anchor semantics as
-        :meth:`analyze` — see that method's docstring for details.
+        ``occurred_at`` and ``extract_with_critique`` carry the same
+        semantics as :meth:`analyze` — see that method's docstring.
         """
 
-        accepted = await self.analyze(text, namespace, occurred_at=occurred_at)
+        accepted = await self.analyze(
+            text,
+            namespace,
+            occurred_at=occurred_at,
+            extract_with_critique=extract_with_critique,
+        )
         completed = await self.wait_for_remember_jobs(accepted.job_ids, opts)
         return AnalyzeWaitResult(
             results=completed.results,
@@ -1360,20 +1381,35 @@ def analyze(
         text: str,
         namespace: Optional[str] = None,
         occurred_at: Optional[Union[str, datetime]] = None,
+        extract_with_critique: bool = False,
     ) -> AnalyzeResult:
         """Synchronous version of :meth:`MemWal.analyze`."""
-        return self._run(self._inner.analyze(text, namespace, occurred_at=occurred_at))
+        return self._run(
+            self._inner.analyze(
+                text,
+                namespace,
+                occurred_at=occurred_at,
+                extract_with_critique=extract_with_critique,
+            )
+        )
 
     def analyze_and_wait(
         self,
         text: str,
         namespace: Optional[str] = None,
         opts: Optional[RememberBulkOptions] = None,
         occurred_at: Optional[Union[str, datetime]] = None,
+        extract_with_critique: bool = False,
     ) -> AnalyzeWaitResult:
         """Synchronous version of :meth:`MemWal.analyze_and_wait`."""
         return self._run(
-            self._inner.analyze_and_wait(text, namespace, opts, occurred_at=occurred_at)
+            self._inner.analyze_and_wait(
+                text,
+                namespace,
+                opts,
+                occurred_at=occurred_at,
+                extract_with_critique=extract_with_critique,
+            )
         )
 
     def embed(self, text: str) -> EmbedResult:

packages/python-sdk-memwal/tests/test_client.py

@@ -592,6 +592,43 @@ async def test_analyze_naive_string_raises(
                 occurred_at="2023-05-25T17:50:00",  # no Z, no offset
             )
 
+    @respx.mock
+    async def test_analyze_with_extract_with_critique_true_sends_flag(
+        self, memwal_client: MemWal
+    ) -> None:
+        """When ``extract_with_critique=True`` the wire body must
+        carry ``extract_with_critique: true`` so the server routes
+        through the two-pass critique path."""
+        mock_seal_session_prereqs()
+        route = respx.post(f"{_TEST_SERVER}/api/analyze").mock(
+            return_value=httpx.Response(200, json={"facts": [], "total": 0, "owner": ""})
+        )
+
+        await memwal_client.analyze(
+            "I moved last Friday",
+            extract_with_critique=True,
+        )
+
+        body = json.loads(route.calls[0].request.content)
+        assert body["extract_with_critique"] is True
+
+    @respx.mock
+    async def test_analyze_with_extract_with_critique_false_omits_flag(
+        self, memwal_client: MemWal
+    ) -> None:
+        """Default ``False`` must omit the field from the wire body so
+        existing callers see byte-identical payloads (the server's
+        ``#[serde(default)]`` resolves the absent field to ``false``)."""
+        mock_seal_session_prereqs()
+        route = respx.post(f"{_TEST_SERVER}/api/analyze").mock(
+            return_value=httpx.Response(200, json={"facts": [], "total": 0, "owner": ""})
+        )
+
+        await memwal_client.analyze("I moved last Friday")
+
+        body = json.loads(route.calls[0].request.content)
+        assert "extract_with_critique" not in body
+
 
 class TestRestore:
     @respx.mock

packages/sdk/src/memwal.ts

@@ -736,6 +736,10 @@ export class MemWal {
         };
         const wireOccurredAt = occurredAtToWire(options.occurredAt);
         if (wireOccurredAt !== undefined) body.occurred_at = wireOccurredAt;
+        // Only include the critique flag when explicitly true; default
+        // off matches the server's `#[serde(default)]` shape and keeps
+        // the wire payload byte-identical for existing callers.
+        if (options.extractWithCritique === true) body.extract_with_critique = true;
         return this.signedRequest<AnalyzeResult>("POST", "/api/analyze", body, [200, 202]);
     }
 

packages/sdk/src/types.ts

@@ -183,6 +183,19 @@ export interface AnalyzeOptions {
      * no server-readable metadata column for it (Architecture A).
      */
     occurredAt?: string | Date;
+    /**
+     * Opt-in two-pass extraction with self-critique. When `true`, the
+     * server runs a second LLM pass that critiques and corrects the
+     * first-pass extracted facts before any fact is embedded or
+     * stored. The critic sees the same `occurredAt` anchor and
+     * related-memories context the first pass saw.
+     *
+     * **Cost:** doubles the per-analyze LLM call count. Default
+     * `false` (or undefined) preserves the single-pass write path —
+     * only flip when the measured benefit justifies the cost. Wire
+     * field is omitted when `false`/`undefined`.
+     */
+    extractWithCritique?: boolean;
 }
 
 /** A fact extracted by analyze() and accepted for background storage. */

services/server/benchmarks/core/client.py

@@ -155,6 +155,7 @@ def analyze(
         text: str,
         namespace: str,
         occurred_at: str | None = None,
+        extract_with_critique: bool = False,
     ) -> AnalyzeResult:
         """
         POST /api/analyze — feed conversation text, extract and store memories.
@@ -175,13 +176,22 @@ def analyze(
         *inside the extracted fact text* — making time-anchored facts
         retrievable. Omit (or pass None) when no anchor is available;
         the server does NOT default to now() — silence is honest.
+
+        `extract_with_critique` flips the server into the
+        two-pass critique path: first pass via the normal
+        extract.v6 extractor, second pass via the critique.v1
+        critic that verifies and corrects the first-pass facts.
+        Doubles the per-analyze LLM call count when set. Default
+        False preserves the single-pass write path.
         """
         body: dict = {
             "text": text,
             "namespace": namespace,
         }
         if occurred_at is not None:
             body["occurred_at"] = occurred_at
+        if extract_with_critique:
+            body["extract_with_critique"] = True
         data = self._post("/api/analyze", body)
         # Response shape varies between server versions:
         # - Synchronous (reference branch): {facts, total, owner}

services/server/benchmarks/run.py

@@ -156,6 +156,17 @@ def stage_ingest(
     start = time.time()
 
     concurrency = config.get("benchmarks", {}).get("concurrency", 10)
+    # when true, every analyze call routes through the two-pass
+    # critique path on the server. Doubles per-analyze LLM cost; we only
+    # flip it for the critique-stack run, not the baseline.
+    extract_with_critique = config.get("benchmarks", {}).get(
+        "extract_with_critique", False
+    )
+    if extract_with_critique:
+        print(
+            "  Two-pass critique extractor ENABLED — expect ~2x "
+            "ingestion wall time vs the single-pass baseline."
+        )
 
     # Build tasks grouped by conversation. Each conversation's chunks are
     # processed SERIALLY (to mirror real-time message ordering and avoid
@@ -222,7 +233,14 @@ def ingest_one_conversation(task):
                 # pass occurred_at (RFC 3339 UTC string or None)
                 # to the server. When present, the server uses it as the
                 # temporal anchor for the extractor prompt.
-                result = client.analyze(text, namespace, occurred_at=occurred_at)
+                # extract_with_critique flips the server into the
+                # two-pass critique path; default False = single-pass.
+                result = client.analyze(
+                    text,
+                    namespace,
+                    occurred_at=occurred_at,
+                    extract_with_critique=extract_with_critique,
+                )
                 memory_ids = [fact.get("id", "") for fact in result.facts if fact.get("id")]
                 if memory_ids:
                     key = session_map_key(conv_id, session_id)
@@ -438,6 +456,12 @@ def process_query(query):
             "mode": mode,
             "judge_model": config.get("judge", {}).get("model", ""),
             "answer_model": config.get("answer", {}).get("model", ""),
+            # pin which write-path the ingest ran through so
+            # later deltas are attributable. Recorded even when False
+            # so baseline runs are explicit about it.
+            "extract_with_critique": config.get("benchmarks", {}).get(
+                "extract_with_critique", False
+            ),
         },
         metrics_overall=_dict_to_category_metrics(overall),
         metrics_by_category={cat: _dict_to_category_metrics(m) for cat, m in by_category.items()},
@@ -574,6 +598,12 @@ def build_parser() -> argparse.ArgumentParser:
     ing = sub.add_parser("ingest", help="Ingest benchmark conversations into Walrus Memory")
     ing.add_argument("benchmark", choices=list(BENCHMARKS.keys()))
     ing.add_argument("--run-id", default=None)
+    ing.add_argument(
+        "--extract-with-critique",
+        action="store_true",
+        help="Route every ingest call through the two-pass critique extractor. "
+        "Doubles per-analyze LLM cost — only flip when measuring the critique stack.",
+    )
 
     # eval
     ev = sub.add_parser("eval", help="Evaluate retrieval with a single preset")
@@ -596,6 +626,12 @@ def build_parser() -> argparse.ArgumentParser:
     full.add_argument("--mode", choices=["retrieval", "e2e"], default="e2e")
     full.add_argument("--run-id", default=None, help="Reuse an existing ingestion")
     full.add_argument("--skip-ingest", action="store_true", help="Skip ingestion stage (assumes run-id already ingested)")
+    full.add_argument(
+        "--extract-with-critique",
+        action="store_true",
+        help="Route every ingest call through the two-pass critique extractor. "
+        "Doubles per-analyze LLM cost — only flip when measuring the critique stack.",
+    )
 
     # report
     rpt = sub.add_parser("report", help="View results")
@@ -641,6 +677,14 @@ def main():
     config = load_config()
     server_cfg = config["server"]
 
+    # Surface CLI flags into the config dict so stage_* helpers (which
+    # don't take args directly) can read them. Only command-level flags
+    # that influence ingestion / eval go here — top-level flags like
+    # --verbose are handled separately.
+    benchmarks_cfg = config.setdefault("benchmarks", {})
+    if getattr(args, "extract_with_critique", False):
+        benchmarks_cfg["extract_with_critique"] = True
+
     client = MemWalClient(
         server_url=server_cfg["url"],
         delegate_key_hex=server_cfg["delegate_key"],
@@ -690,7 +734,11 @@ def main():
                 "metadata. Upgrade the server to a build that exposes these fields."
             )
             sys.exit(1)
-        print(f"  prompt versions: extract={prompt_versions['extract']} ask={prompt_versions['ask']}")
+        critique_ver = prompt_versions.get("critique", "n/a")
+        print(
+            f"  prompt versions: extract={prompt_versions['extract']} "
+            f"critique={critique_ver} ask={prompt_versions['ask']}"
+        )
         # Stash on `config` so stage_eval picks it up without a signature change.
         config["_server_prompt_versions"] = prompt_versions
     except SystemExit:

services/server/src/routes/admin.rs

@@ -129,6 +129,8 @@ pub async fn health(State(state): State<Arc<AppState>>) -> Json<HealthResponse>
         // and ask (`/api/ask`) — no separate config to drift.
         prompt_versions: PromptVersions {
             extract: crate::services::extractor::FACT_EXTRACTION_PROMPT_VERSION.to_string(),
+            critique: crate::services::extractor::FACT_EXTRACTION_CRITIQUE_PROMPT_VERSION
+                .to_string(),
             ask: ASK_SYSTEM_PROMPT_VERSION.to_string(),
         },
     })

services/server/src/routes/analyze.rs

@@ -324,10 +324,24 @@ pub async fn analyze(
     // references ("last Friday") to absolute dates *inside the extracted
     // fact text* (Architecture A — no metadata column, date flows into
     // the encrypted blob + embedding only).
-    let extracted = state
-        .extractor
-        .extract_with_context(&body.text, &related_texts, body.occurred_at)
-        .await?;
+    //
+    // when `body.extract_with_critique` is true, route through
+    // the two-pass critique path instead. The critique pass internally
+    // reuses `extract_with_context` for the first pass, so the same
+    // `related_texts` + `occurred_at` context applies to both passes —
+    // the critic sees the same anchor the extractor did. ~2x LLM cost
+    // when enabled; default false preserves the single-pass behaviour.
+    let extracted = if body.extract_with_critique {
+        state
+            .extractor
+            .extract_with_critique(&body.text, &related_texts, body.occurred_at)
+            .await?
+    } else {
+        state
+            .extractor
+            .extract_with_context(&body.text, &related_texts, body.occurred_at)
+            .await?
+    };
     let raw_fact_count = extracted.raw_count;
     let facts = extracted.facts;
     let reserved_additional_weight = rate_limit::analyze_additional_weight(facts.len());