Skip to content

IngestPipelineService

Defined in: packages/memory-ltm/src/ingest/ingest-pipeline.service.ts:20

Stream B0 — Typed Ingest Pipeline (13 steps).

This service owns the ordered execution of pipeline steps.

  • runSyncSteps(): runs steps 1–6 synchronously in the tool response path.
  • Step 7 (PostgresWrite) and steps 11–12 (EmbeddingGenerate / SearchIndexUpdate) remain in MemoryLtmService so they can participate in the existing transaction and error handling.
  • runAsyncHooks(): fires steps 3, 8–10, and 13 as fire-and-forget no-ops today, to be wired in Streams F, I, and D respectively.

new IngestPipelineService(privacyFilter, topicDetector): IngestPipelineService

Defined in: packages/memory-ltm/src/ingest/ingest-pipeline.service.ts:26

PrivacyFilterStep

TopicDetectorStep

IngestPipelineService

computeHash(content): string

Defined in: packages/memory-ltm/src/ingest/ingest-pipeline.service.ts:111

SHA-256 of trimmed, lower-cased content.

string

string


runAsyncHooks(ctx, memoryId): void

Defined in: packages/memory-ltm/src/ingest/ingest-pipeline.service.ts:85

Fire-and-forget async hooks for steps 8–10 and 13.

These are no-ops today, wired in their respective streams:

  • step 8: EventLogAppend (Stream I)
  • step 9: EntityGraphUpdate (Stream F)
  • step 10: BacklinkCompute (Stream F)
  • step 13: VaultSync (Stream D)

Call this after a successful PostgresWrite (step 7).

IngestContext

string

void


runSyncSteps(ctx): Promise<IngestContext>

Defined in: packages/memory-ltm/src/ingest/ingest-pipeline.service.ts:51

Run pre-write synchronous steps (1–6) on the context.

Returns the enriched context, or a context with aborted: true if an exact duplicate was detected via content hash.

The caller (MemoryLtmService) is responsible for:

  • step 7 (PostgresWrite)
  • step 11 (EmbeddingGenerate)
  • step 12 (SearchIndexUpdate)

After step 7 succeeds the caller should invoke runAsyncHooks() for steps 8–10 and 13.

IngestContext

Promise<IngestContext>