Activate served graph firehose ingest

[aaltshuler]

Stack

Depends on #438. This PR is intentionally based on codex/graph-native-stream-ingest so the review contains only F7a activation.

Summary

• expose graph-first POST /graphs/{graph_id}/stream/ingest plus served-only omnigraph stream ingest
• gate body ownership with a strong graph-authority ETag and stream ordered, redacted NDJSON outcomes
• reuse the existing lazy enrollment, Lance MemWAL, resident fold driver, recovery, cancellation, and bounded result ownership; add no coordinator, storage format, or recovery strand
• flush low-rate same-declaration runs on a non-resetting 50 ms coalescing boundary
• preserve open-mode and authenticated default-deny policy semantics before body polling, and redact private startup authority evidence
• update OpenAPI, release notes, operator/user docs, invariants, and the test ownership map

Validation

• cargo check --workspace --locked
• focused engine graph-NDJSON, cancellation, low-rate liveness, and non-resetting deadline tests
• enabled-runtime server end-to-end ingest and query visibility
• authenticated default-deny, bearer-auth, media/runtime refusal, and startup-redaction tests
• CLI stream client tests, including exact 200 NDJSON enforcement and no body replay
• full server OpenAPI suite: 86 passed
• scripts/check-agents-md.sh
• git diff --check

The hour-long full workspace test gate was intentionally not rerun locally.

---

Greptile Summary

The PR activates graph-native served streaming ingest across the engine, HTTP API, CLI, shared wire types, OpenAPI contract, and documentation.

• Adds strong graph-authority ETag preflight before request-body ownership.
• Streams ordered, redacted NDJSON outcomes while reusing existing MemWAL enrollment, folding, recovery, and cancellation machinery.
• Adds served-only CLI ingest support and updates policy, lifecycle, liveness, integration, and OpenAPI coverage.

Confidence Score: 5/5

The PR appears safe to merge because no blocking failure remains within the eligible follow-up-review scope.

No blocking failure remains.

Important Files Changed

Filename	Overview
crates/omnigraph/src/db/omnigraph/stream_ndjson.rs	Implements bounded graph-native NDJSON parsing and ordered logical outcome production without an accepted review finding.
crates/omnigraph/src/db/omnigraph/stream_request.rs	Adds graph-authority preflight and request ownership for the served streaming bridge without an accepted review finding.
crates/omnigraph/src/db/omnigraph/stream_ingest.rs	Integrates graph-level streaming requests with existing lane enrollment, MemWAL, and fold authority.
crates/omnigraph-server/src/handlers.rs	Adds the authenticated and policy-gated NDJSON endpoint, strong ETag challenge, and streaming response transport.
crates/omnigraph-server/src/lib.rs	Registers the new route and OpenAPI operation and wires its HTTP error mappings.
crates/omnigraph-cli/src/client.rs	Adds one-shot preflight, strong ETag validation, non-replayed streaming request bodies, and exact NDJSON response handling.
crates/omnigraph-api-types/src/lib.rs	Defines the public challenge and per-line graph-logical stream result contracts.
Cargo.lock	Adds the CLI's direct futures dependency while leaving reported advisory-bearing transitive versions unchanged.

Sequence Diagram

sequenceDiagram
    participant CLI as CLI / HTTP client
    participant Server as Omnigraph server
    participant Policy as Cedar policy
    participant Engine as Graph stream ingest
    participant WAL as Lance MemWAL
    participant Driver as Resident fold driver
    participant Manifest as Graph manifest

    CLI->>Server: "POST /graphs/{id}/stream/ingest (no If-Match)"
    Server->>Policy: Authorize stream_ingest
    Policy-->>Server: Allow
    Server->>Engine: Effect-free authority preflight
    Engine-->>Server: Current graph authority token
    Server-->>CLI: 428 + strong ETag, no body polling
    CLI->>Server: Retry with If-Match + NDJSON stream
    Server->>Engine: Start checked graph ingest
    Engine->>WAL: Lazily enroll and durably append rows
    WAL-->>Engine: Per-row durable outcome
    Engine-->>CLI: Ordered redacted NDJSON outcomes
    Driver->>WAL: Coalesced fold
    Driver->>Manifest: Atomically publish visible graph state

Loading

_{Reviews (4): Last reviewed commit: "fix(ci): scope AWS stack floor to tests" | Re-trigger Greptile}

Context used (5)

• Knowledge Base — OmniGraph core storage: manifest, table store, and Lance access
• Knowledge Base — OmniGraph write path
• Knowledge Base — Server Request Handling
• Knowledge Base — Policy enforcement (Cedar)
• Knowledge Base — CLI and Config

[devin-ai-integration]

Devin Review found 1 potential issue.

View 1 additional finding in Devin Review.

[devin-ai-integration]

🟡 Firehose clients are told recovery is pending when the real problem is an unconfirmed write

An ingest request whose durability acknowledgement came back unknown is reported to the caller as "graph recovery must complete before stream ingest can begin" (StatusCode::SERVICE_UNAVAILABLE arm at crates/omnigraph-server/src/lib.rs:1100-1102), so operators chase a recovery problem that does not exist.
Impact: Operators debugging a failed stream start are pointed at the wrong subsystem and may wait for or trigger recovery work that is unrelated to the actual failure.

How the ambiguous-acknowledgement branch collapses into the recovery message

ApiError::from_graph_stream_start (crates/omnigraph-server/src/lib.rs:1079-1113) deliberately rewrites AckUnknown from 500 to 503 (crates/omnigraph-server/src/lib.rs:1084-1093) so the ambiguity stays retryable/unavailable. The subsequent message table then keys purely off the HTTP status, so both OmniError::RecoveryRequired and OmniError::AckUnknown land on the same 503 string. The new test graph_stream_start_errors_redact_private_engine_evidence only asserts that private evidence is absent, so the mismatch is not caught. The acknowledgement_unknown flag is already in scope and can select a distinct, still-redacted message.

Suggested change

	StatusCode::SERVICE_UNAVAILABLE => {
	"graph recovery must complete before stream ingest can begin"
	}
	StatusCode::SERVICE_UNAVAILABLE if acknowledgement_unknown => {
	"graph stream ingest durability acknowledgement is unknown; retry the request"
	}
	StatusCode::SERVICE_UNAVAILABLE => {
	"graph recovery must complete before stream ingest can begin"
	}

---

Was this helpful? React with 👍 or 👎 to provide feedback.