consume-nats-message

Skill: Consume NATS messages

Purpose

Enable collaborating projects to "correctly consume messages"—pull pending messages in batches via NATS MCP tools according to the producer contract, process each one with the Tolerant Reader and ack correctly. Default drain-style: a single call pulls until the queue is empty or the limit is reached; a single failure does not block the whole batch, and an aggregated result is finally returned.

Core Objective

Primary objective: drain the pending messages of the subject corresponding to the producer contract, decoding each one with the Tolerant Reader → business processing → correct ack / nak / term, and output aggregated statistics.

Success criteria：

✅ Project cache .cortex/nats.yaml has been read
✅ Producer contract has been located (local vendored preferred) and passed the contract_version compatibility check;when the contract is missing, Bootstrap has generated a draft and it has been confirmed by the user
✅ Pull consumer established / reused (durable name named according to the cache prefix)
✅ Loop fetch until drained / cap_reached / idle_timeout one of
✅ Each message: Tolerant Reader decode → business processing → ack decision correctly executed
✅ A single failure does not interrupt the whole batch, and is written to the failure list
✅ Return an aggregated receipt (counts + failure details + exit reason + resume suggestions)

Acceptance test: when the consumer processes a batch containing 1 message with a newly added optional field, it does not throw an error and acks correctly; when it contains 1 message with an unrecoverable business error, term + DLQ delivery; when it contains 1 message with a temporary failure, nak so the broker redelivers.

Scope boundaries

This skill is responsible for：

Reading the project cache + loading the producer contract
Bootstrap when the contract is missing: peek samples + infer schema + generate a draft (for consumer use only, not sent externally)
Establishing / reusing a JetStream pull consumer via NATS MCP
Drain loop + per-message processing + per-message ack decision
Tolerant Reader decoding
Failure tolerance (a single message does not block the whole batch) + DLQ delivery (term path)
Aggregated receipt

This skill is not responsible for：

Modifying the producer contract → the consumer does not write producer files; the draft generated by Bootstrap is only a local consumer draft, not the producer's authoritative contract
Promoting the draft contract to active → must be manually version-bumped by the user after review and alignment with the producer owner
Creating broker resources (stream / durable) → IaC responsibility (if missing for the first time, report an error pointing to the IaC owner)
Actual business processing logic → the consumer repo's own code / handler; this Skill only coordinates messages and acks
Long continuous push subscription → not in the Skill invocation model (application-layer worker responsibility)

Prerequisites

NATS MCP server already connected (providing mcp__nats__jetstream_pull_consumer / mcp__nats__ack / mcp__nats__nak / mcp__nats__term / mcp__nats__publish and other tools; the specific tool names are subject to the connected server)
The current working directory is the consumer repo
A frozen snapshot of the producer contract has been vendored into the consumer repo (recommended) or an accessible contract path—it may be missing on first use; the skill will automatically enter Phase 2.5 Bootstrap to generate a draft contract

Execution flow

Phase 0: Announce

"I am using the consume-nats-message skill to drain a batch of pending messages."

Phase 1: Read project cache

Read .cortex/nats.yaml(consumer repo root directory)
Extract fields:broker_url / service_source / durable_name_prefix / dlq_handler_dir / vendor_contracts_dir / consume_defaults
User parameters can override consume_defaults（max_messages / batch_size / fetch_timeout / idle_threshold）
Cache missing or field missing → guide completion and prompt to commit

Phase 2: Locate producer contract

By priority:

<vendor_contracts_dir>/<producer>/<event>-contract.md(frozen snapshot, including @version anchoring)
Contract path or URL explicitly provided by the user (including @version）
both missing → enter Phase 2.5 Bootstrap(do not terminate)

Read contract:

subject (including MAJOR version segment, e.g. clarification.session.requested.v1）
Required headers list
payload field table
QoS / DLQ subject / max_deliver / backoff

Phase 2.5: Bootstrap (contract draft guidance, executed only on first run/contract missing)

Purpose: A missing contract is normal in the early stage of collaboration. This phase generates a draft contract via "read-only probing + inference", so the user can start reasonably even with zero prerequisites.Throughout, do not ack, do not change consumer offset, do not modify broker state。

Preconditions：

Triggered only when no contract is found in Phase 2; skip this phase when a contract is found in Phase 2
The output of this phase is a contract draft for status: draft , before the user reviews and confirmsdo not enter the formal drain of Phase 5
The draft is for the bootstrap consumer's own use only;do nottreat it as an external authority, and do not replace the producer's SSOT contract

Steps：

List broker topology(peek, do not subscribe)
- List all streams / subjects and message counts
- Priority display: streams matching the producer name provided by the user (e.g. producer=zentao → ZENTAO stream) come first
- If the user does not provide a producer name, list in descending order of message count
Select target subject
- User provided producer + event → use <producer>.<event>.* pattern matching
- Only producer provided → list all subjects under that stream for the user to choose
- Neither provided → list all subjects + counts + most recent sample for the user to choose
Peek sample messages(do not create durable, do not ack)
- Use stream get or an ephemeral consumer to directly read the most recent N messages (default N=5, can be overridden by bootstrap_sample_size )
- Across samples, merge the header key set and payload field set, and record each field's type and occurrence rate (basis for required/optional inference)
Infer schema
- subject pattern: extract the version segment from sample subjects (e.g. .v1 / .v2 suffix)
- headers: those with 100% occurrence rate are classified as "required", <100% as "optional"; standard NATS headers (Nats-Msg-Id / Nats-Stream etc.) are not included in inference
- payload: same as above for JSON fields, nested objects are processed recursively; for enum values, list all distinct values in the samples
- QoS / max_deliver / DLQ: cannot be inferred in this phase, leave TBD
Generate draft contract
- Path:<vendor_contracts_dir>/<producer>/<event>-contract.md
- frontmatter must include:
```
contract_version: 0.1.0
status: draft
inferred_from: bootstrap-peek
inferred_at: <ISO 8601>
inferred_sample_size: <N>
inferred_sample_seq_range: <first>-<last>
authoritative: false
```
- Body title pre-warning banner:
  
  ⚠️ This contract was inferred and generated by the consume-nats-message Bootstrap mode from N sample messages,non-authoritative. Please contact the producer owner to confirm field semantics, QoS, and DLQ rules, then status promote to active、authoritative promote to true。
- .lock write to <contract-name>@0.1.0
Request user confirmation
- Show to user: draft path + headers inference table + payload inference table + fields to be clarified (marked TBD）
- Question:
  - (a) Continue formal drain: from stage 3 onward, process messages according to the draft contract (with ack)
  - (b) Look only, do not act: terminate this execution and let the user first align the contract with the producer owner
- Default with no input → choose (b), to avoid mistakenly acking under the draft contract

Version compatibility in Bootstrap mode：

contract_version: 0.1.0 is treated as pre-stable; stage 3 skips MAJOR/MINOR compatibility checks
Each time the user confirms a draft upgrade, manually bump to 1.0.0 and mark status: active

Stage 3: Version compatibility check

Read contract frontmatter contract_version
and compare with the version already aligned by the consumer itself (<vendor_contracts_dir>/<producer>/.lock record <contract-name>@<version>）
mismatch: indicate version drift
- MINOR/PATCH drift: Tolerant Reader is usually compatible; warn and continue
- MAJOR drift (new subject): the subject the consumer actually subscribes to has changed → terminate and prompt to upgrade the vendor snapshot
No .lock recorded: first alignment, write .lock then continue

Stage 4: Establish / reuse pull consumer

Via MCP NATS tools:

durable name: <durable_name_prefix>-<event-slug>
If durable does not exist → error pointing to IaC (violates spec §5.5.2 "do not create resources at application code runtime")
If it exists → attach directly
ack policy: as agreed in contract (default explicit）
max_deliver: as agreed in contract

Stage 5: Drain loop

processed = 0
last_non_empty_at = now
while True:
    if processed >= max_messages:
        exit_reason = "cap_reached"; break
    batch = fetch(batch_size, fetch_timeout)
    if batch is empty:
        if now - last_non_empty_at >= idle_threshold:
            exit_reason = "drained"; break
        continue
    last_non_empty_at = now
    for msg in batch:
        try:
            process_one(msg)        # 阶段 6
        except Exception as e:
            record_failure(msg, e)  # 不中断整批
        processed += 1

Exit reason values:drained / cap_reached / idle_timeout(consecutive empty fetches + limit not reached).

Stage 6: Process one by one (Tolerant Reader)

For each message:

Deduplication: check the application-layer deduplication set (by Nats-Msg-Id，TTL ≥ broker duplicate_window); on hit → ack directly and skip
Decode headers: read required headers; missing Nats-Msg-Id / X-Source / X-Type → treat as violating message (business-unrecoverable, take the term path); ignore unknown headers
Decode payload：
- Missing required field → business-unrecoverable → term + DLQ
- Unknown field → ignore without error
- Unknown enum value → take the fallback branch (business-defined)
OTel continuation: extract Traceparent continue span (if applicable)
Call business handler: consumer repo's own processing logic; the handler return result determines the ack path
Decide ack: proceed to stage 7

Stage 7: Ack decision

Business result	MCP tool	Side effect
Success	`mcp__nats__ack`	broker confirms consumption
Retryable errors (transient network / downstream rate limiting)	`mcp__nats__nak`	broker redelivers per backoff; eventually reaches max_deliver and automatically terms
Non-recoverable errors (schema violation / permanent business failure)	`mcp__nats__term` + `mcp__nats__publish` Deliver to DLQ	No further redelivery; the original message is copied to `<original>.dlq` with failure reason header

DLQ message headers appended:

X-DLQ-Original-Subject: original subject
X-DLQ-Reason: failure summary
X-DLQ-Failed-At: ISO 8601 timestamp

Phase 8: Aggregation Receipt

Output:

exit_reason: drained                    # drained / cap_reached / idle_timeout
duration: 12.3s
counts:
  fetched: 87
  acked: 80
  naked: 4
  termed: 3
  dlq_sent: 3
  duplicates_skipped: 1
failures:
  - msg_id: 01HX...A1
    subject: clarification.session.requested.v1
    decision: term
    reason: missing required payload field 'sessionId'
  - msg_id: 01HX...A2
    subject: ...
    decision: nak
    reason: downstream timeout (retry will be attempted)
backlog_hint: |
  exit_reason=drained，无需续跑。
  （若为 cap_reached，建议再次调用本 Skill 续 drain。）

Error Handling

Situation	Handling
MCP NATS server not connected	Prompt to check MCP configuration
`.cortex/nats.yaml` missing	Guide to fill in
Contract missing / not vendored	Enter Phase 2.5 Bootstrap—peek samples + inference + generate draft,do notterminate directly
After Bootstrap inference, user selects (b) view only, do not modify	Terminate after draft is persisted; the next call to Phase 2 will hit it
Durable consumer does not exist	Error points to IaC owner (violates spec §5.5.2)
MAJOR version drift	Terminate and prompt to upgrade the vendor snapshot
Single-message decode failure	term + DLQ + write failure list,do not interrupt the whole batch
Single-message business retryable failure	nak, broker redelivers naturally
Fetch timeout (no available messages on broker side)	Count toward empty fetch counter, trigger idle_threshold exit

Anti-patterns

❌ Throw an error on a single failure and interrupt the whole batch — must try/catch per message, record it in the failure list, and continue
❌ Throw an error on decoding to an unknown field — Tolerant Reader requires ignoring it
❌ Not acking on successful processing — the message will be redelivered by the broker, causing idempotent false-death
❌ Using nak for non-recoverable business failures — it will keep redelivering until max_deliver; should directly term + DLQ
❌ DLQ delivery without the original headers / failure reason — loses operational visibility
❌ Modifying the producer contract — the consumer does not write producer files; use backlog to track upgrades
❌ A continuous push subscription occupying the Skill invocation — the Skill is a one-shot invocation; implement long-running application-layer workers separately
❌ Terminating directly when the contract is missing — should enter Bootstrap peek + inference draft (without acking or changing offset), so the first-use threshold is not too high
❌ The Bootstrap-inferred draft is not marked status: draft / authoritative: false — it will be mistaken for an authoritative contract and referenced externally, polluting the SSOT
❌ Creating a durable consumer / executing ack during Bootstrap — must be read-only peek to avoid polluting broker state

Relationship with other assets

Structural contract ruler：specs/nats-messaging.md—authority for subject / headers / payload / QoS / Tolerant Reader / DLQ rules
Contract document skeleton：specs/cross-team-contract.md—follow its version-number anchoring conventions when vendoring upstream contracts
Production-side companion：publish-nats-message—the other end of the same contract
broker behavior authority：https://docs.nats.io——NATS pull consumer / ack policy / Services API are subject to the official documentation