Add DOCUMENT_OCR_MODE=sync|batch (default sync). In batch mode the tier-3 OCR
processor submits documents to the embedding gateway's async Batch OCR routes
(POST /v1/ocr/batch + GET /v1/ocr/batch/{job_id}, astrolabe-cloud-website#372)
for ~50% cheaper large-corpus backfill. The direct Mistral OCR path is left
untouched. Tracked on Deck #332.
Batch jobs run minutes-hours, so the OCR tier cannot block (the procrastinate
worker reclaims jobs in `doing` after INGEST_STALLED_JOB_SECONDS). Instead it
submits, records the gateway job id in a new per-tenant `batch_ocr_jobs` table
(procrastinate args are immutable across retries), and raises a BatchPending
signal that TieredEscalationStrategy turns into a same-queue deferred re-poll —
releasing the worker slot between polls. On completion the per-page markdown is
indexed like the sync path; a failure or a job past
DOCUMENT_OCR_BATCH_MAX_WAIT_SECONDS marks the document parse-failed.
Batch is opt-in and gateway-only: with the direct mistral backend, no gateway
URL, or the inline/memory pipeline (which can't defer), it falls back to sync.
One batch job per document (coalescing N docs/job is a follow-up).
- embedding/gateway_batch_client.py: submit/poll client (reuses GatewayTokenProvider).
- vector/batch_ocr_store.py + migration 008: job tracking (portable SQLite+PG).
- document_processors/escalation.py: BatchPending control-flow signal.
- document_processors/ocr.py: batch state machine + sync fallback.
- vector/processor.py: thread doc identity to the OCR tier; raise BatchPending
from the pending sentinel; propagate it as control flow (not a failure).
- vector/queue/procrastinate.py: BatchPending -> same-queue retry_in, exempt
from the transient cap (bounded by the processor's deadline).
- config + docs; tests across client/store/processor/strategy/parse-tier.
1653 unit tests pass; ruff + ty green.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
A stall is often systemic (a Qdrant/embedding outage stalls every in-flight
job), so reclaiming the whole batch at now() every */5min tick would
thundering-herd a recovering dependency, bypassing TieredEscalationStrategy's
per-job backoff. reclaim_stalled_ingest_jobs now offsets retry_at by a fixed
delay (INGEST_RECLAIM_RETRY_DELAY_SECONDS, default 30s; 0 = legacy immediate).
Also document the hot-vs-restart flag asymmetry: INGEST_ESCALATION_ENABLED is
re-read per job; INGEST_TRANSIENT_MAX_ATTEMPTS is snapshotted at worker startup.
Deck #323.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
- metrics: update_ingest_queue_depth guarded on `not by_queue`, which conflated
None (memory backend no-op) with {} (postgres, ALL queues drained). When every
queue drains at once, get_ingest_job_counts_by_queue returns {} and the
pre-zero loop was skipped, leaving a stale ghost backlog in the gauge. Guard on
`by_queue is None` only; add an all-drained regression test.
- procrastinate: note that INGEST_TRANSIENT_MAX_ATTEMPTS is snapshotted at
blueprint-build time (restart to pick up changes).
Deck #323.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
- metrics: update_ingest_queue_depth now pre-zeroes every managed ingest queue
before applying live counts, so a queue that drains to empty (and drops out of
procrastinate's list_queues_async) reads 0 instead of sticking at its last
non-zero value (ghost backlog in Grafana/alerts). Adds a regression test.
- procrastinate: comment that _is_transient_infra_error treats all qdrant errors
as transient deliberately (bounded same-tier retry; over-broad is acceptable).
- escalation: note next_tier is the building block; production routing uses
ProcessorRegistry.next_available_tier.
- tests: add evaluate_escalation fast+ocr-only low-confidence -> ocr case.
Deck #323.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
- Register the periodic stalled-job reclaim on a dedicated ingest-maintenance
queue that every worker drains (any --tier), so reclaim still fires when the
fast fleet is scaled to zero and only ocr workers run. procrastinate's
periodic-defer dedup keeps it single-run across drainers.
- escalation: mark `unsupported`/`forced` reason labels as reserved (not raised).
- processor: note that options/progress_callback are intentionally not threaded
through _parse_pdf_tier yet (symmetric with the inline path).
- tests: assert TieredEscalationStrategy backoff progression (4/8/16/…/300s);
cover get_ingest_pending per-queue aggregation + the legacy job_counts
fallback; add an external-path zero-page no-escalation case; use the canonical
INGEST_QUEUE_FAST instead of the back-compat alias.
Deck #323.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Split external (procrastinate) document processing into per-tier queues so a
document is attempted at most once per tier and requeued to the next tier's
queue on a low-quality parse, using procrastinate's native retry.
- escalation.py: TIER_LADDER (fast->structured->ocr) + EscalateError signal
- registry: process_tier (one tier) + evaluate_escalation post-parse gate
(reuses classify_from_text) + next_available_tier; shared _classify_result
and _oversize_result with the inline pipeline
- processor: process_document(tier=...) runs one tier and raises EscalateError
before embed (junk text never indexed); inline memory path unchanged
- queue/procrastinate: ingest-fast|structured|ocr queues; TieredEscalationStrategy
(queue-hop on EscalateError, bounded same-tier transient retry); queue-aware
task; producer defers to ingest-fast; per-queue counts + all-queue reclaim
- cli: worker --tier {fast,structured,ocr}
- billing: pages_ocr usage event + pipeline_tier metadata (paid OCR billed apart)
- observability: astrolabe_ingest_queue_depth{queue,status} gauge + per-queue
counts in nc_get_vector_sync_status / management status endpoint
- config: INGEST_ESCALATION_ENABLED (default true), INGEST_TRANSIENT_MAX_ATTEMPTS
INGEST_ESCALATION_ENABLED=false and INGEST_QUEUE=memory preserve prior behaviour.
Deck #323.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
- _clear_vector_sync_state also nulls shutdown_event / scanner_wake_event on
shutdown, symmetric with the stream/producer fields (the next startup rebinds
them via _wire_vector_sync_state).
- Comment that the "DocumentTask" string subscript in LocalTransport is
intentional (TYPE_CHECKING-only class; anyio ignores the runtime type arg).
- Move app.py's annotation-only IngestTransport / TaskProducer imports under
TYPE_CHECKING (the module uses `from __future__ import annotations`), keeping
only build_transport at runtime.
Refs: Deck #196
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
- Clear the module-singleton ingest references (task_producer,
document_send_stream, document_receive_stream) on lifespan shutdown via a new
_clear_vector_sync_state() helper, mirroring the eviction_task_group cleanup.
Defense-in-depth so a late webhook (or a module-singleton integration test)
can't touch a producer/stream backed by an already-closed resource.
- Add IngestTransport.backend_name ("memory"/"postgres") and use it in both
lifespan log lines, removing the last settings.ingest_queue read from the
background-sync setup — the lifespan no longer inspects the backend at all.
- Cover backend_name in the build_transport adapter-selection tests.
Refs: Deck #196
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
- LocalTransport.run_consumers increments active_consumer_count per worker
(instead of once after the loop) so the count is accurate if a later
tg.start() raises mid-pool.
- Add LocalTransport.aclose() to explicitly close its owned send/receive stream
ends (belt-and-suspenders against unclosed-resource warnings; anyio aclose is
idempotent, and by shutdown the scanner is already winding down). Reworded the
base IngestTransport.aclose() docstring to point at the overrides.
- Inline ingest_transport.producer at the scanner/user_manager call sites,
dropping the single-use task_producer alias in both lifespan paths.
- Annotate DistributedTransport._producer explicitly as ProcrastinateTaskProducer
so the drain() coupling is visible and ty catches drift.
- Add a unit test for LocalTransport.aclose() (closes the owned streams,
idempotent).
Refs: Deck #196
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
- Rename the lifespan-local `transport` to `ingest_transport` in both paths so it
no longer shadows the get_app(transport=...) HTTP-transport parameter.
- Log the memory backend selection in build_transport, symmetric with the
postgres branch, so startup logs name the chosen ingest backend either way.
- Note in _wire_vector_sync_state why eviction_task_group is intentionally not
set there (it only exists once the lifespan's task group is running).
Refs: Deck #196
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
- Add IngestTransport.active_consumer_count (0 by default; LocalTransport stores
the started count) so app.py logs the worker count without re-checking
INGEST_QUEUE — the last backend-knowledge leak in the lifespan is gone.
- Document that DistributedTransport is postgres/procrastinate-specific by design
(aclose() calls ProcrastinateTaskProducer.drain()); other distributed backends
would be separate IngestTransport subclasses.
- Clarify the _wire_vector_sync_state log line (writes app.state + singleton, not
only the singleton).
- Strengthen the LocalTransport test: assert active_consumer_count transitions
0→N and that each worker receives a distinct cloned receive stream.
Refs: Deck #196
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Finish the hexagonal ports-&-adapters split started in #183. The producer side
already had a TaskProducer port + adapters, but the consumer side was
unabstracted and the INGEST_QUEUE selection leaked into a duplicated
`if use_postgres:` branch across both app.py lifespan paths.
Introduce an IngestTransport ABC (vector/queue/transport.py) that bundles the
producer with running (or not running) the in-process consumer pool, built by a
single build_transport() factory:
- LocalTransport (INGEST_QUEUE=memory): in-process anyio stream drained by an
N-worker pool that run_consumers starts.
- DistributedTransport (INGEST_QUEUE=postgres): wraps ProcrastinateTaskProducer;
run_consumers is a no-op because the consumer is the external `worker` role.
Both lifespan paths now call build_transport + _wire_vector_sync_state (new
helper that centralizes the app.state / module-singleton / browser-app writes) +
transport.run_consumers + transport.aclose(), with no INGEST_QUEUE branching and
no getattr drain probe. Adding a future backend (Redis/NATS/SQS) is one new
adapter + one build_transport arm, with no app.py or scanner change.
Preserves the single-tenant parallelism invariant (one shared multiplexed queue
+ N-worker pool, per-document not per-user dispatch) and documents it in
ADR-028. The worker CLI is unchanged (it is the external consumer).
Refs: Deck #196 (Deck #197 tracks the explicit parallelism regression test)
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Round-4 review (non-blocking) items:
- get_procrastinate_conninfo: warn on an empty connect_timeout= value (it falls
back to the 10s default); preserve an explicit connect_timeout=0.
- Document the _doc_queueing_lock user_id invariant (NC rejects ':' in usernames).
- docs/configuration.md: note that `db downgrade` leaves procrastinate's tables
in place and how to drop them on a full teardown.
- reclaim_stalled_ingest_jobs: debug heartbeat log when nothing is stalled.
- Drop the redundant list() wrap in the integration stalled-jobs assertion.
Logging pattern: define a module-level `logger = logging.getLogger(__name__)`
and use it instead of function-local or inline getLogger(__name__) calls
(config.py, config_validators.py, tests/.../test_scope_authorization.py). The
test file's dev-only `scripts.*` import gets a ty: ignore since it resolves via
sys.path at runtime, not as an installed package.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
🟡 Document the _doc_queueing_lock ":" delimiter invariant (user_id and the
controlled doc_type enum are colon-free, so the key is collision-safe; a
future doc_type with ":" must not be added).
🟡 API pod no longer opens the procrastinate connector twice on startup: add
ProcrastinateTaskProducer.ensure_schema() (applies the schema on the
already-open pool) and have both lifespan branches build the producer then
ensure_schema — one open/close cycle, matching the worker. build_producer now
returns the concrete producer type.
🟢 Document in ports.py that a long-lived-connection producer may optionally
provide drain() (lifespan probes via getattr).
🟢 Add a unit test that a non-credential pipeline error propagates (for
procrastinate's RetryStrategy) and still closes the client via finally.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
🟡 Document why ProcrastinateTaskProducer.connect() uses `await app.open_async()`
(AwaitableContext: await opens a long-lived pool, closed by drain()) and add a
connect()/drain() lifecycle unit test (InMemoryConnector) asserting the pool is
opened by connect and closed by drain — previously untested.
🟡 get_procrastinate_conninfo: forward connect_timeout from DATABASE_URL or
default 10s so an unreachable DB can't hang worker/API startup indefinitely;
warn only on other dropped query params. + tests.
🟢 INGEST_DELETE_SUCCEEDED_JOBS (default true) makes the worker's succeeded-job
deletion configurable for audit retention.
🟢 Worker startup logs via logger.info (structured/OTel) instead of click.echo.
🟢 INGEST_STALLED_JOB_SECONDS (default 300) makes the crash-reclaim threshold
tunable for slow embedding backends; reclaim reads it per-run.
The broad `except` in _apply_ingest_queue_schema_open is kept deliberately:
procrastinate wraps psycopg errors, so narrowing to psycopg.errors.* would miss
the wrapped DDL-conflict and turn a benign concurrent-apply race into a failure;
the presence re-check re-raises genuine errors.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
🔴 nc_get_vector_sync_status reported pending=0 for INGEST_QUEUE=postgres: the
AppContext/OAuthAppContext per-session yields snapshotted the stream fields but
never forwarded task_producer, so lifespan_ctx.task_producer was always None.
Convert task_producer to a @property that reads _vector_sync_state live (like
eviction_task_group), removing the snapshot field so the yields can't drop it.
Add a regression test pinning the contract on both contexts.
🟡 Remove the unused _RECLAIM_TASK_NAME constant.
🟡 get_procrastinate_conninfo: warn + document that DATABASE_URL query params
(application_name, connect_timeout, …) are dropped.
🟡 worker: open the procrastinate App once — apply_ingest_queue_schema gains
manage_connection=False so the worker reuses its own open connector instead
of a redundant open/close before run_worker_async.
🟢 Clarify the apply-schema broad-except comment (non-race errors re-raise) and
document the deliberate Any typing in ingest_status.get_ingest_pending.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Re-architect document ingest from the shared NATS-glued document-processor to a
per-tenant, in-process model owned by nextcloud-mcp-server (Deck #183). The MCP
server now owns both sides of ingest:
- Producer (api role): the scanner defers one job per changed document into the
app's Postgres via procrastinate (queueing_lock dedup; no execution lock, so a
crashed worker can't deadlock a doc — Qdrant upserts are idempotent).
- Consumer (worker role): `nextcloud-mcp-server worker` drains the queue and runs
the existing process_document pipeline; a periodic task reclaims jobs orphaned
in `doing` by a crash.
INGEST_QUEUE selects the transport (auto: postgres when DATABASE_URL is Postgres,
else the in-process anyio queue for SQLite/dev). procrastinate manages its own
tables (applied on a fresh DB at startup and by `db upgrade`). The vector-sync
status surface reads job counts from Postgres in postgres mode. procrastinate +
psycopg3 ship in the [postgres] extra; the app's own engine still uses asyncpg
(driver unification is a follow-up handled in the rendered Helm chart).
NATS JetStream, the Postgres-queue stub, the bus status subscriber, and nats-py
are removed.
BREAKING CHANGE: the external-NATS-ingest env vars are removed
(INGEST_MODE, STATUS_BACKEND, INGEST_BUS_URL, INGEST_BUS_NUM_REPLICAS,
FACT_EVENT_EMITTER). Use INGEST_QUEUE (memory|postgres) and the `worker`
command instead. TENANT_ID is retained (no longer NATS-subject-charset-validated).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Follow-up to PR #814 review.
NatsStatusSubscriber.run() called task_status.started() *after* the fallible
pull_subscribe, so a NATS broker that wasn't ready when the MCP server started
would crash the lifespan instead of retrying. Bus status is a non-critical
observability path, so:
- signal started() before the first subscribe (semantics: "loop is running",
not "subscription succeeded");
- retry a failed subscribe with backoff instead of propagating;
- on a real fetch error (not an idle timeout) drop the subscription and
re-subscribe rather than fetching against a possibly-dead handle.
Also anchor the _content_hash etag-threading TODO to the PR #814 review thread
so it is discoverable outside git blame.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
SonarCloud's python:S7632 parses the literal ``# NOSONAR`` token wherever it
appears — including inside explanatory comments that *quote* the directive —
and treats the following text as a malformed suppression. The actual bare
``# NOSONAR`` suppression lines are fine; the flagged lines were the prose
comments describing them. Reword those comments to drop the inner ``#`` so the
analyzer no longer sees a directive.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The new decomposition modules used `# NOSONAR: reason` (colon form), which
SonarCloud flags as a malformed suppression comment (python:S7632) and which
fails to suppress the intended issue. Switch to the repo's bare `# NOSONAR`
convention with the rationale in a comment above, matching config.py and
auth/storage.py. This also lets the suppression silence python:S7503 (async
method without await) on the protocol-required no-op aclose stubs.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
- gateway_client: guard token cache with a lazy anyio.Lock so concurrent
embed calls share one M2M token request instead of racing
- status subscriber: distinguish idle fetch timeouts from real broker
errors (log + 5s backoff) instead of swallowing all and spinning
- nats: warn when the bus URL uses unencrypted transport (non-tls://)
- collection_metadata: accept an optional shared httpx client, make TLS
verify explicit, document the unauthenticated control-plane contract
- replace python -O-stripped asserts with explicit ValueError in the bus
status builder and the api metadata source
- document why the nil-UUID sentinel point can't collide with content ids
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
SonarCloud:
- Resolve 6 S5332 hotspots (http→https in test fixture URLs).
- S6418: hoist the unauthenticated AsyncOpenAI placeholder to a named constant
+ NOSONAR (genuine non-secret; gateway ignores it when unauthenticated).
- Fix two reliability bugs: None-index guard in the gateway token-cache test
(S2259) and float `> 0.0` instead of `!= 0.0` in the sentinel test (S1244).
- status.py idle path sleeps 0.1s instead of sleep(0) (S7491); NOSONAR on the
protocol-required async no-await aclose() stubs (S7503).
Claude review:
- Remove three leftover debug print() calls in app.py (logger.info already
covers them).
- payload_backfill: drop parsed_at from the backfilled-keys docstring (it is
per-document state, not a deployment scalar); add a clean 404 precondition
for BasicAuth deployments without an OAuth token verifier.
- status.py: task_status typed TaskStatus | None (drop type: ignore).
- nats.py: TODO to thread etags for file/deck/news; note etag default → None.
- factory: warn on unknown INGEST_BUS_URL scheme; raise ValueError instead of
assert for the external-mode preconditions.
- docs/configuration.md: document the decomposition hook-point env vars + that
nats-py ships core (lazy-imported) and external+bus uses two NATS connections.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Adds the seven §10.2 hook-point modules + five env vars so Astrolabe Cloud can
offload document processing to the external document-processor / embedding
gateway. Purely additive: with every setting unset the server behaves exactly
as today, so self-hosters are unaffected (Deck #92).
Hook points (all default to current monolith behavior):
- config: EMBEDDING_PROVIDER, INGEST_MODE, STATUS_BACKEND,
COLLECTION_METADATA_SOURCE, FACT_EVENT_EMITTER (+ supporting settings),
validated in Settings.__post_init__ (fail-fast STATUS_BACKEND=local with
INGEST_MODE=external); shared canonical.py.
- vector/payload_keys.py + acl_hash.py: cross-impl NAMESPACE/point_id (§2.2)
and BLAKE2b-128 ACL hash (§11), pinned by fixtures shared with the
document-processor repo.
- embedding/gateway_client.py: OpenAI-compatible GatewayProvider authenticating
via M2M OIDC client-credentials (separate realm); manual-only registry entry.
- vector/collection_metadata.py: sentinel-point / API metadata source with env
fallback.
- vector/queue/: hexagonal ingest producer ports + memory/NATS adapters
(Postgres seam); INGEST_MODE=external publishes mcp.ingest.requested.{tenant}
instead of the in-memory stream and skips the in-process processor pool. The
lifespan becomes a composition root across both deployment branches.
- vector/queue/status.py: STATUS_BACKEND=bus subscriber feeding a StatusStore
the vector-sync status endpoint reads.
- admin/payload_backfill.py: POST /api/v1/admin/payload-backfill (admin scope);
processor writes the new payload keys; query-side ACL pre-filter gated behind
ACL_PREFILTER_ENABLED (default off).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
