Deal Context Generation Workflow

Last updated at 2026-06-15 13:14 AEST.

A high-level walkthrough of how the platform turns a user's "analyze this asset" action into a structured, versioned deal context — the single source of truth that every downstream capability (solution strategy, proposal generation, evaluation) reads from.

At a glance

A handful of numbers capture the shape of the workflow — how it's structured, how much it can read, and how it weighs what it finds:

Overview

Deal context generation is the platform's ingestion engine: it takes a single uploaded deal asset (an RFP, SOW, meeting transcript, BOM, quote, environment document, email thread, etc.), reads and understands it with LLM agents, and folds what it learns into a structured, versioned DealContext record for the deal. The characteristics below are grounded in the mechanisms documented in the rest of this file — they describe what the code actually does, not aspirations.

The remainder of this document walks the flow end to end, from the triggering mutation through to where the finished deal context is stored.

1. Entry Point

The user triggers deal context generation from the web app (e.g. after uploading or selecting an asset on a deal), which calls a single GraphQL mutation on the API:

• Mutation: initiateDealContextGeneration

This is a "fire-and-forget" mutation: it returns almost immediately with a job handle (AsyncProcessingJob). The actual analysis runs asynchronously in the background. The frontend tracks progress through GraphQL subscriptions.

2. Input Required

The mutation accepts a small InitiateDealContextGenerationInput:

Field	Required	Purpose
assetId	Yes	The asset (uploaded file) to analyze.
dealId	Yes	The deal the asset belongs to and is analyzed for.

The logged-in user is taken from the auth context — the resolver never trusts a caller-supplied user or organization ID. Both fields are typed as GraphQL ID and trimmed at the boundary via TrimPipe; the substantive validation is the resolver's existence + org-scope checks below (the DTO itself carries no field-level class-validator rules).

Preconditions

Before the resolver enqueues anything, it validates:

1. The deal exists and belongs to the caller's organization (deal.findFirstOrThrow scoped by organizationId).
2. The deal input exists — i.e. the asset is actually linked to the deal (dealInput.findFirstOrThrow on { assetId, dealId }).

If either check fails, the mutation throws and nothing is queued.

The "kickoff"

This kickoff is intentionally lightweight (no placeholder output rows, no advisory lock, no deal-stage advance — those belong to proposal generation). The resolver:

1. Touches DealInput.updatedAtUtc so the asset's "last analyzed" ordering reflects this request.
2. Creates an AsyncProcessingJob of entity type DealInput and job type GenerateDealContext, with a nested DealAsyncProcessingJob row linking dealId + assetId. The job starts in Pending.
3. Publishes the Pub/Sub message (section 3) and returns the job to the UI.

No real-time event is published at kickoff. The first asyncProcessingJobChanged (status → InProgress) is published later by the subscriber, the instant it atomically claims the job (section 4) — so the UI's job indicator is driven by the actual claim, not the enqueue.

3. The Pub/Sub Message

The resolver publishes a message to a Google Cloud Pub/Sub topic. This is the hand-off from the synchronous request to the in-process background listener.

• Topic: <environment>.deal-context.generate (the literal name is environment-prefixed and resolved per environment via env vars; e.g. api-1ke-local-au.deal-context.generate).
• Schema: assetProcessingPubSubMessageDataSchema
• Ordering key: when ordering is enabled, the message is published with key deals/<dealId> so Pub/Sub delivers same-deal messages one at a time (reinforcing the per-deal serialization in section 4).

Payload:

{
  "correlationId": "<asyncProcessingJobId>",
  "payload": {
    "assetId": "...",
    "asyncProcessingJobId": "...",
    "userId": "...",                 // who triggered the run
    "userProvidedContext": "..."     // optional free-text; threaded into the agent prompts when present
  }
}

The message is tiny — just enough pointers for the subscriber to find everything it needs in the database. No business data is carried on the wire.

4. How the Subscriber Processes the Message

The subscriber is DealContextGenerationSubscriberService. It runs in-process inside the API service (started on module init) as a Pub/Sub streaming-pull listener on the topic above; the number of messages it processes concurrently and the ack-deadline / lease-extension windows are bounded by env vars. The listener runs on every API instance, so Pub/Sub load-balances messages across the instances currently running.

For each message, it does the following:

1. Validate the message envelope. If the shape is wrong, ack and drop it — it would just fail forever on replay.
2. Load the job pointers. Read the DealAsyncProcessingJob to get the asset (contentHash, filePath, originalFileName), the job (id, organizationId, status), and the dealId. If the job is already Cancelled or Completed, return silently (the message acks).
3. Claim the job atomically via DealContextGenerationJobClaimerService (see "Per-deal serialization" below). This flips Pending/Failed → InProgress under a per-deal lock, recovers a stale sibling if one is wedged, or nacks for redelivery if a healthy sibling is genuinely running.
4. Register a shutdown handler so an unexpected Cloud Run shutdown reverts the job back to Pending rather than leaving it stuck InProgress. (Released on every exit path.)
5. Initialize or restore WorkflowProgress plus one WorkflowStep row per phase, walking the canonical runtime order (extract-asset-content → analyze-asset-requirements → persist-extracted-data → generate-deal-context). On retry, only the not-yet-completed steps are reset and currentStepIndex is moved to the last completed step.
6. Seed the Mastra request context with the asset content hash, original file name, the retry flag, and handles to the services the steps need (Google Cloud Storage, hashing, Redis, Redis pub/sub) plus the workflow-progress / step-id map — Mastra steps can't read Nest DI directly, so they read these from the request context at run time.
7. Start the Mastra workflow run (section 6) with the initial state (assetId, dealId, organizationId, userId, optional userProvidedContext).
8. On success, flip the job to Completed (unless it was cancelled/completed out from under the run). All persistence of results happens inside the workflow steps (section 7), not here.

Per-deal serialization (the job claimer)

Deal context generation must run one-at-a-time per deal because each run reads the previous version's context. The claimer runs a single interactive transaction:

1. SELECT id FROM "Deal" WHERE id = <dealId> FOR UPDATE — a per-deal row lock that serializes all concurrent claim attempts for the deal (held for milliseconds, not the multi-minute run).
2. Look for another GenerateDealContext job on the same deal that is InProgress.
3. If one exists and its startedAtUtc is within the stale threshold, throw GooglePubSubConcurrentJobError → the message nacks and is redelivered later. If it's older than the threshold (or null), fail it inside the same transaction (stale-sibling recovery) so this message can proceed.
4. CAS-claim our own row: updateMany from { status in [Pending, Failed] } → InProgress. If zero rows update, the row was claimed/cancelled elsewhere — throw the same concurrent-job error.

Status-changed events (for both a recovered stale sibling and the newly claimed job) are published over Redis pub/sub after the transaction commits.

Failure handling

• Deterministic failures (schema violations, contract bugs — surfaced as DeterministicWorkflowError) are marked failed and the pub/sub message is acked, because retrying would just reproduce the same error.
• Transient failures (network, provider blip, DB disconnect, OOM) revert the job to Pending (avoiding a UI flicker to "Failed") and leave the message nacked. Pub/Sub redelivers up to the configured retry budget, then routes it to a dead-letter topic.
• Job cancellation is checked before claiming and again after the run; a cancelled job exits cleanly without flipping to failed.

5. Data Retrieved by the Subscriber & Steps

Deal context generation has no single up-front "snapshot" compile like proposal generation; instead each step reads exactly what it needs:

Source	What's read
DealAsyncProcessingJob	The asset (contentHash, filePath, originalFileName), organizationId, dealId.
Asset	filePath, fileSize, originalFileName — to choose a content-extraction path and fetch bytes from GCS.
DealContext (latest)	The deal's most recent context version — both for grounding the analysis and as the merge base.
DealInput	sourceType and createdAtUtc — drive the composite merge weight (source reliability + recency).
Deal identity	Customer name / deal name, compiled into a "deal identity" system message that grounds extraction and synthesis in what the deal is actually about.
Redis cache (content-hashed)	Prior asset-analysis and per-chunk results, so identical bytes skip the LLM extraction.

The actual asset bytes (or extracted text) are fetched from the organization's private Google Cloud Storage bucket inside the step, on whichever extraction path matches the file type (section 6).

6. How Data Flows Through the Workflow

The workflow itself is a Mastra workflow (dealContextGenerationWorkflow) composed of two sequential Mastra steps that surface to the UI as four progress phases. There is no parallel fan-out at the step level (each phase depends on the previous one); the only concurrency is within the analysis phase, where a large asset is chunked and the chunks are analyzed in parallel under a concurrency cap.

Step graph (top to bottom)

Each phase publishes a live progress message as it transitions, so the user always sees where the work is. The four messages map one-to-one to the four phases:

1. "Extracting content from asset…" — extract-asset-content
2. "Extracting metadata and requirements…" — analyze-asset-requirements
3. "Saving extracted requirements…" — persist-extracted-data
4. "Generating deal context…" — generate-deal-context

What each phase does

Extract asset content — picks one of three content paths by MIME type (resolved from the filename) and size, fetching bytes from GCS only inside the path's own frame so large buffers stay GC-eligible:

Size backstops on extracted text (all paths): an extraction exceeding MAX_EXTRACTED_CONTENT_CHARS (10 MB of characters — a signature of malformed extraction, e.g. inlined images) fails deterministically rather than churning. Otherwise a conservative ~3 chars/token estimate decides between a single inline call and a chunked sweep (below).

Analyze asset requirements — calls the asset-analysis agent with the extracted content (plus the deal-identity grounding and any latest-context / user-provided context). The agent returns a single structured assetAnalysis: a suitability verdict, source-type classification, metadata (vendor/tech mentions, environment hints, current-state hints, commercial hints, compliance mentions, business-objective hints, section hints), and zero or more structured requirements (each with category, priority, acceptance criteria, tech constraints, provenance, and a forced self-justification that it represents a real solution obligation — items that can't argue themselves as a genuine solution obligation, including legal/contractual boilerplate, are dropped). Most assets are analyzed in a single agent call; an asset too large to fit the model's context window is split into chunks, analyzed in parallel under a pLimit(2) cap, and reconciled into one result — see Chunking below for the full mechanics.

Persist extracted data — writes the extracted requirements as first-class rows (publishing a dealRequirementChanged event and an activity), records the classified source type on the DealInput, and writes a deterministic per-asset contribution summary (facet counts shown on the deal's assets card). On a cache hit the agent doesn't run, so a synthetic Generation row (modelId: 'cache-hit') is created to attribute the persisted requirements.

Generate deal context — calls the deal-context agent to merge the new asset's metadata and requirements into the deal's latest context, producing a new structured DealContext value (customer, objectives, constraints, current state, environment, engagement mix, commercials, NFRs, sizing, risks, selection, stakeholders, timeline, provenance, …). The merge is governed by the composite-weight guidance block (below). The result is validated against dealContextSchema, a deterministic completeness check is run (advisory), authoritative IDs are stamped by code, and a new immutable version is written (section 7). Short-circuit: if the upstream suitability verdict was an explicit isSuitableForIngestion: false, this phase is skipped entirely — no DealContext row is created and the job closes cleanly.

Chunking

Most assets are analyzed in a single agent call. Chunking is the fallback for the rare document that is too large to analyze in one pass — and it changes how the analysis is produced, so it's worth understanding end to end.

Why chunk at all. An LLM can only read so much input in one call — its context window. The asset-analysis model's usable input budget, after reserving room for its own instructions and its answer, is ~762k tokens (MAX_CONTENT_TOKEN_BUDGET). Send more than that and the model rejects it or silently drops the overflow, producing a broken or partial analysis. So the extract phase measures the text up front and routes accordingly:

• Fits the budget (the common case — clears it by a wide margin for ordinary documents) → one whole-document agent call.
• Exceeds the budget → split into chunks, each sized to fit, analyzed separately.

How big a chunk is. Each chunk is capped at the same budget that triggers chunking: maxSize ≈ 762k tokens, which at the code's conservative 3-chars-per-token estimate is ~2.3 million characters (chunking.ts). Chunks are therefore enormous — a document only chunks when it is larger than that, so a chunked document is millions of characters (hundreds of pages). An individual requirement's text — a few hundred characters — is minuscule next to a single chunk.

Each chunk gets its own full LLM analysis. A chunk is sent through the exact same asset-analysis agent call as a small whole document would be, and produces a complete result for its slice of the text: a suitability verdict, a source-type classification, metadata / high-level signals, and structured requirements. Each chunk's result is cached individually (keyed on (contentHash, chunkHash)), so a retry never re-pays for chunks that already succeeded.

Merging the chunk results — two tracks. Once every chunk is analyzed, the results are combined into one assetAnalysis, but the two kinds of output are merged differently:

1. Metadata & signals → merged by plain code, no LLM. The lists (vendor mentions, environment hints, compliance mentions, …) are concatenated and exact-match deduplicated, confidence scores are averaged, and suitability is OR'd across chunks (suitable if any chunk found it suitable). This is deterministic: the same chunk results always merge to the same output.
2. Requirements & source type → a second LLM pass, when warranted. Chunking creates two new problems: the same requirement may appear in two chunks, and a chunk may have mis-guessed the document type from just its fragment. So the asset-analysis-reconciliation agent (frontier-tier) pools all candidate requirements across chunks, settles the single document source type by reconciling the per-chunk votes, and re-judges every candidate, dropping duplicates and anything that isn't a genuine solution obligation. Two caveats: if the chunks produced zero requirements, this call is skipped (the source type is settled by a deterministic majority vote instead); and if the call fails transiently, the code keeps all candidates and falls back to the deterministic vote — a timeout must never silently delete requirements.

Does splitting hurt analysis accuracy? A boundary could in principle fall between two related pieces of text. In practice this is heavily mitigated:

• Splitting is structure-aware, not a blind character cut. The chunker uses MDocument.fromMarkdown(...) with a recursive strategy and zero overlap. Recursive splitting prefers the largest semantic separators first (markdown headings → paragraphs → sentences → raw characters only as a last resort) and cuts at the highest-level boundary that keeps a chunk under the size cap — so a split lands at a section or paragraph seam, not mid-sentence.
• The chunk size makes a mid-requirement cut near-impossible. Boundaries occur roughly every ~2.3M characters, at structural seams; the odds one lands inside a specific few-hundred-character requirement are vanishingly small.
• Reconciliation re-judges everything together. Even if a boundary clipped something, all candidates are pooled and re-judged at the end, so a duplicate or weak fragment tends to get dropped or consolidated.

The honest residual limitation: there is no chunk overlap (the usual belt-and-suspenders fix for straddling content). The pipeline bets instead on huge chunks + structural splitting + reconciliation. The real blind spot isn't a requirement being physically cut — it's context genuinely spread across a boundary (the rationale in one chunk, the requirement statement in the next): each chunk analyzed without the full picture, and reconciliation can only work with what each chunk independently produced.

How reliable is the deterministic dedup? It is exact-match dedup (dedupeStrings / dedupeObjects in merge-asset-analysis-results.helpers.ts): strings are trimmed and collapsed via a Set; objects collapse only when their serialized JSON is byte-identical. It reliably removes exact duplicates but not semantic ones — "AWS" and "aws" both survive (trim, not lowercase), as do "Microsoft Azure" vs "Azure", or two SKU objects differing by one field. That's intentional: the metadata/signals are hints that the downstream deal-context merge agent consolidates semantically anyway, so a few redundant hints are low-harm. Requirements, where duplicate quality actually matters because they become first-class persisted rows, get the LLM reconciliation pass instead of this exact-match dedup.

The composite-weight merge contract

When an existing deal context is present, the step computes a per-asset composite weight and embeds it, with an explicit resolution contract, into the agent's system prompt:

• Composite weight = 0.5 × sourceTypeWeight + 0.3 × extractionConfidence + 0.2 × recencyFactor, where recency decays with a 120-day half-life (using the asset's issue date if known, else its upload date).
• Source-type weights rank document trustworthiness: RFP / SOW 0.90; BOM / compliance doc / environment doc / quote / sizing spreadsheet 0.85; email thread 0.70; meeting transcript 0.60; CRM record / other 0.50.
• Resolution rules told to the agent: append-and-deduplicate lists; for conflicting scalars prefer the higher-weighted value; when two values' weights are within a close threshold (0.1), keep both with provenance annotations.

RFP / SOW 0.90

BOM / compliance / environment / quote / sizing 0.85

Email thread 0.70

Meeting transcript 0.60

CRM record / other 0.50

This keeps merges principled and deterministic in spirit — a recent RFP wins over a months-old transcript — rather than letting the model arbitrate conflicts ad hoc.

Caching, retries, and progress

• Every phase writes a WorkflowStep row with started, completed, or failed timestamps — this powers the live progress bar the frontend shows ("Extracting content from asset…", "Generating deal context…").
• Content-hash caching: the asset-analysis result and each chunk result are cached in Redis keyed on the asset's content hash. A redelivery (or any asset with byte-identical content) skips extraction + analysis and goes straight to persisting. There is no shouldForceRegeneration flag — the content hash naturally keys the cache, and a changed asset produces a different hash.
• Phase short-circuit on retry: the assetAnalysisStep only short-circuits when all three of its sub-phases previously completed; if any one (e.g. persist) failed, it re-runs from the cache-hit / agent path so the failed sub-phase is re-bracketed cleanly. The generate-deal-context phase short-circuits independently when its row already completed.
• Version-collision retry: DealContext versions are unique per deal; if a concurrent write grabs the next version number first (a P2002 unique violation), the step re-fetches the latest version, bumps to +1, and retries the create.

7. Where the Generated Deal Context Is Stored

Unlike proposal generation, the output is not a file — it's structured rows in Postgres. Nothing is uploaded to GCS by this workflow (GCS is only read, for the source asset). The two persistence points are:

During the persist-extracted-data phase:

1. Requirement rows — the extracted, sanitized requirements are written as Requirement rows (sourceType = Asset) and auto-accepted onto the deal via DealRequirement rows. Each is paired with an ExtractedRequirement row that attributes it to the producing Generation (with a confidence score, the original text, and the model's reason) and an AssetExtractedRequirement row that carries provenance back to the source asset (and section). A dealRequirementChanged event and a RequirementsExtracted deal activity are published.
2. DealInput.sourceType — the asset's classified source type (RFP, SOW, transcript, …).
3. DealInput.contributionSummary — a deterministic snapshot of what this asset contributed (facet counts: requirements, commercial signals, current-state signals, compliance mentions, constraints, stakeholders, section hints), shown on the deal's assets card.

During the generate-deal-context phase:

1. A new immutable DealContext row with:
   • The next sequential version for the deal (with the version-collision retry above).
   • trigger = AssetAnalysis, assetId, createdByUserId, organizationId, and the producing generationId — full lineage of which asset and which agent call produced this version.
   • The structured context in value (JSON). Code owns the authoritative IDs — dealId, organizationId, version, and the suitability assetId are written deterministically from workflow state after the agent merge, so a model-hallucinated value can never win. The value blob's version is kept in lockstep with the column.
   • An advisory completeness block (constraints ≥ 2, environment inventory ≥ 4, provenance facts ≥ 8, selection ≥ 1, timeline ≥ 3); violations are logged but never block.
2. A dealContextChanged (CREATED) event is published over Redis pub/sub, scoped to the organization, so every open client tab sees the new version immediately.

Every agent call along the way writes a Generation audit row (system prompt, user prompt, model id, token usage, response-time, and a response snapshot), and failed runs mark their in-flight Generation rows failed with structured error details.

8. Security

The flow ingests customer-supplied documents and feeds them to LLM agents, so security applies at the API boundary, on the wire, in the agents, and in storage.

• Identity is never caller-supplied. The triggering mutation derives the logged-in user — and therefore the organization — from the authenticated request context. Every database read and write in the kickoff, the claimer, and the workflow steps is scoped to that resolved organizationId.
• Input is checked at the boundary. InitiateDealContextGenerationInput's string fields are trimmed via TrimPipe, and the resolver verifies the deal (existence + org scope) and the deal-input link before anything is queued — a request for a deal/asset the caller can't see is rejected synchronously.
• The Pub/Sub message carries no business data. Only pointer IDs cross the wire (section 3); the subscriber re-reads everything from the database under the job's organization scope, so there is no customer content in the payload to leak or tamper with.
• Agent output is treated as untrusted and schema-bound. The asset-analysis agent's structured output runs with prompt-injection guarding enabled (jsonPromptInjection), and every agent result — asset analysis, reconciled requirements, and the deal context itself — is parsed against its Zod schema (the single source of truth) before it is persisted. The authoritative identity fields on the deal context are overwritten by code regardless of what the model emits.
• Generated data lives under the organization's scope. DealContext, Requirement / DealRequirement / ExtractedRequirement, and Generation rows all carry organizationId; source assets live in the organization's private GCS bucket and are only read, never re-exposed by this flow.
• Real-time events stay within the organization. The asyncProcessingJobChanged / dealRequirementChanged / dealContextChanged events are fanned out only to clients in the publishing organization, following the platform's subscription-payload conventions (relations stripped, sensitive scalars redacted).

9. Performance & Cost Controls

Deal context generation runs on every uploaded asset, so the design keeps wall-clock latency, memory, and LLM spend bounded.

• Asynchronous offload. The analysis runs in a background Pub/Sub listener rather than under the triggering request (sections 3–4), so no HTTP request is held open and a slow run never ties up request capacity.
• Content-hash caching. Asset-analysis and per-chunk results are cached in Redis by content hash, so a redelivery — or a re-upload of identical bytes — skips the LLM extraction entirely and goes straight to persistence.
• Phase-level short-circuit on retry. A retry re-runs only the phases that hadn't completed (tracked on WorkflowStep), not the whole pipeline.
• Bounded agent calls. Every agent stream is wrapped in a stall-guard with a ceiling timeout (13 minutes) so a stalled provider stream can't wedge the run. The ceiling is held below the recovery sweep's stale window so a genuinely wedged run fails the workflow before the sweep would race it.
• Bounded fan-out and memory. Large assets are chunked against a budget sized to the smallest frontier input window (~762k tokens after output/prompt reserves), with a MAX_EXTRACTED_CONTENT_CHARS (10 MB) deterministic-fail backstop ahead of it; chunk analysis runs under a pLimit(2) concurrency cap because each in-flight reasoning-mode stream holds an SDK buffer plus the model's response in memory. Content-extraction paths deliberately fetch GCS buffers inside the narrowest possible frame so the bytes are GC-eligible before persistence runs. Native (PDF/image) handling is capped at 3 MB before falling back to the Docling text-extraction path.
• Tiered models and reasoning budgets. The asset-analysis, deal-context, and asset-analysis-reconciliation agents all run frontier-tier with a high reasoning budget (24,576 thinking tokens) for the hard synthesis and judgment work. The reconciliation agent is only invoked when a chunked asset actually produced requirements to merge (an empty set settles the source type deterministically from the per-chunk votes and skips the call entirely).
• Suitability short-circuit. An asset judged unsuitable for ingestion skips the expensive deal-context synthesis agent altogether rather than burning its full wall-clock ceiling producing low-signal output.
• Self-healing without manual ops. A periodic + on-startup recovery sweep reaps jobs stuck Pending or InProgress past their stale thresholds — failing them via an idempotent compare-and-swap so they surface for retry or remediation instead of sitting stuck — and a shutdown handler reverts in-flight jobs to Pending on Cloud Run scale-in, so a wedged or interrupted run becomes retryable on its own.

Summary

End-to-end the flow looks like: