diff --git a/docs/adr/0061-per-dimension-partial-accept.md b/docs/adr/0061-per-dimension-partial-accept.md new file mode 100644 index 000000000..e39ef439f --- /dev/null +++ b/docs/adr/0061-per-dimension-partial-accept.md @@ -0,0 +1,167 @@ +--- +status: accepted +date: 2026-08-12 +deciders: Allen D. Householder +consulted: Vultron protocol maintainers +informed: Vultron contributors +--- + +# Adjudicate Received `ParticipantStatus` Per Dimension, Not as a Unit + +## Context and Problem Statement + +A `ParticipantStatus` is not a single state value. It is a snapshot of several +independent state machines: `rm` (Report Management), `vfd` (vendor fix path), +`em` (Embargo Management), `pxa` (public state) and `consent` (Participant +Embargo Consent). When one arrives over the wire in +`Add(ParticipantStatus, CaseParticipant)`, the receiving CaseActor must decide +what to believe. + +Before this decision it decided all-or-nothing. `ValidateRMTransitionNode` +refused any backwards `rm` step and any status for a participant already at +terminal `RM.CLOSED`, and its FAILURE inside the `AppendParticipantStatusBT` +Sequence discarded the entire snapshot — including dimensions the receiver had +no grounds to refuse. Worse, the FAILURE aborted the enclosing Sequence before +`StatusUpdateGuard` and `EmitAddCaseStatusToSelfNode`, so the Seam 1 → Seam 2 +emit never happened and embargo teardown silently did not run (ADR-0046, +RSH-01-003, RSH-01-004). A vendor that had closed its report management +workflow could not report deploying its fix at all. + +Reported as ISSUE-2235, under the liberal-accept epic ISSUE-2229 (Postel's +maxim: be conservative in what you send, liberal in what you accept). + +## Decision Drivers + +- The dimensions are genuinely independent state machines; a refusal in one + carries no information about the others. +- Liberal accept (ISSUE-2229): refuse the narrowest thing that must be refused. +- Refusals must be *visible*. The pre-existing behaviour was a silent drop. +- CLP-10-006 receive-side ordering: precondition guards run before + `GuardedCommit`; guards MUST NOT write to the DataLayer. +- The canonical `CaseLedgerEntry` is hash-chained and replicated to every + participant. Whatever it snapshots becomes every replica's view. +- Monotonic visibility: a replica must never un-see progress it has observed. + +## Considered Options + +- **Keep all-or-nothing, but return SUCCESS on refusal.** Fixes the aborted + Seam 2 emit only. +- **Per-dimension adjudication with a filtered snapshot.** Refuse each + dimension independently; carry the participant's current value forward for + the refused ones; record the resulting filtered `ParticipantStatus`. +- **Per-dimension adjudication plus an outbound refusal message.** As above, + plus a new wire message telling the sender what was refused. + +## Decision Outcome + +Chosen option: **per-dimension adjudication with a filtered snapshot**. + +`FilterParticipantStatusDimensionsNode` +(`vultron/core/behaviors/status/nodes/dimension_filter.py`) adjudicates `rm`, +`vfd` and `pxa` separately, then publishes a filtered `ParticipantStatus` in +which each refused dimension carries the participant's current value forward. +That filtered object is what gets persisted, appended to the participant, and +snapshotted in the canonical ledger entry. It runs as a read-only precondition +guard of `add_participant_status_tree`, replacing +`CheckParticipantRMNotClosedNode`. + +Per-dimension rules: + +- `rm` — accepted when it confirms the current value, is a valid adjacent + transition, or is a monotone forward jump. `RM.CLOSED` is terminal: once a + participant has closed, no further `rm` value is accepted, *including* + `CLOSED` again. +- `vfd` and `pxa` — each is a triple of independent one-way latches + (`v→V`, `f→F`, `d→D`; `p→P`, `x→X`, `a→A`). Accepted when no component + regresses from uppercase back to lowercase. +- `em` — not adjudicated here. Embargo state is Seam 2's (`add_case_status_tree`, + RSH-02-001); Seam 1 adjudicating it would duplicate and could contradict + Seam 2's decision. Tracked in ISSUE-2256. + +The refusal is made visible through the canonical ledger rather than a new wire +message: the committed entry snapshots the accepted portion, so it differs from +what the sender asserted and every participant sees the receiver's actual view. +No new message type is introduced (an outbound refusal is deferred; see +ISSUE-2255 for the separate problem that the HTTP response is `202 Accepted` +regardless of outcome). + +A status update whose accepted portion is indistinguishable from the +participant's current state is refused *in full* — nothing appended, no ledger +entry committed. Such an assertion carries no acceptable information, and +recording it would grow both the status history and the hash chain with no +state change. + +Symmetrically on the replica side, +`ApplyParticipantStatusFromLedgerNode` now ratchets `rm`: an +`Announce(CaseLedgerEntry)` that would move the local `rm` backwards on the +progress scale has that dimension carried forward at the local value, while +every other dimension is applied as the entry describes it. Lateral moves at +the same rank (`VALID` ↔ `INVALID`, `DEFERRED` ↔ `ACCEPTED`) are the Case +Actor re-adjudicating, not a regression, and are applied unchanged. + +### Consequences + +- Good, because a refused dimension no longer destroys accepted state, and no + longer kills the Seam 1 → Seam 2 emit or embargo teardown. +- Good, because the canonical ledger — the thing that actually replicates — now + records what the receiver believes rather than what the sender claimed. +- Good, because the guard is read-only with respect to the DataLayer, so it + fits CLP-10-006 ordering and can run before the commit. +- Good, because a `RM.CLOSED` participant can still report VFD/PXA progress. +- Bad, because the sender still gets no explicit signal that a dimension was + refused; it must observe the canonical ledger. The HTTP-status half of that + gap is ISSUE-2255. +- Bad, because the guard-to-append handoff uses the py_trees blackboard, which + is process-global and not cleared between executions. Both keys are therefore + written on every tick (with `None` when inapplicable) and matched by object ID + on read. This is a real hazard, not a hypothetical one. +- Neutral, because `CheckParticipantRMNotClosedNode` remains in the codebase, + marked deprecated with a pointer to its replacement, rather than being + removed. + +## Validation + +`test/core/behaviors/status/test_partial_accept_participant_status.py` covers +each rule: a refused `rm` with accepted `vfd`/`pxa`, survival of the Seam 2 +emit, the ledger snapshot carrying the accepted `rm`, a `RM.CLOSED` participant +advancing `vfd`, whole-update refusal committing no entry, and the replica-side +RM ratchet. + +## Pros and Cons of the Options + +### Keep all-or-nothing, but return SUCCESS on refusal + +- Good, because it is a one-line change that unblocks embargo teardown. +- Bad, because accepted dimensions are still silently discarded — the + liberal-accept violation remains. +- Bad, because the canonical entry would still snapshot the raw assertion. + +### Per-dimension adjudication with a filtered snapshot + +- Good, because it matches the actual structure of the data: independent state + machines adjudicated independently. +- Good, because it makes the refusal visible in the one artifact that is + replicated and hash-chained. +- Neutral, because it needs a guard→append handoff channel; the blackboard is + the available mechanism and carries the leakage hazard noted above. + +### Per-dimension adjudication plus an outbound refusal message + +- Good, because the sender learns immediately and precisely what was refused. +- Bad, because it introduces a new wire message type and its authorization + semantics — a much larger surface than the bug requires. +- Bad, because a refusal message invites refusal loops between peers that + disagree; that needs its own design. + +## More Information + +- ISSUE-2235 — the bug this decision resolves. +- ISSUE-2229 — liberal-accept epic (Postel's maxim). +- ISSUE-2255 — receive path returns `202 Accepted` regardless of BT outcome. +- ISSUE-2256 — Seam 2 `em` adjudication. +- ADR-0046 — two-seam authorization model. +- `notes/sync-ledger-replication.md` — monotonic visibility and the + reject-on-divergence invariants. + +Generated spec requirements: `received-status-handling.yaml` RSH-05-001 through +RSH-05-008. diff --git a/docs/adr/index.md b/docs/adr/index.md index 7e9627ebb..a6d214d4b 100644 --- a/docs/adr/index.md +++ b/docs/adr/index.md @@ -129,6 +129,7 @@ General information about architectural decision records is available at These jobs fail intermittently due to inter-container HTTP delivery timeouts -> (async race windows). Root cause documented in `plan/incoming/learnings/` -> entry `20260731-async-race-windows-in-fv-demo.md`. When a new occurrence is -> confirmed, `pr-execute` will open or comment on a `flaky-test` + `bug` issue -> and record it here. +| `fcvcv Demo Integration` | #2233 (was #2216, closed) | 2026-08-13 | +| `fvcv-handoff Demo Integration` | #2233 (was #2216, closed) | 2026-08-13 | +| `fvcv-handoff Invariant Harness` | #2233 (was #2216, closed) | 2026-08-13 | +| `fcvcv Invariant Harness` | #2233 | 2026-08-13 | +| `fcv-reject Invariant Harness` | #2233 (was #2121, closed) | 2026-08-13 | +| `fv Invariant Harness` | #2233 | 2026-08-13 | + +> Some of these jobs fail intermittently due to inter-container HTTP delivery +> timeouts (async race windows). Root cause documented in +> `plan/incoming/learnings/` entry `20260731-async-race-windows-in-fv-demo.md`. +> When a new occurrence is confirmed, `pr-execute` will open or comment on a +> `flaky-test` + `bug` issue and record it here. +> +> **The six rows pointing at #2233 are not flaky** — they fail on *every* run +> until the engage-case 422 lands, and they are listed here only because +> `pr-execute`'s dedup procedure looks here first. The Demo Integration pair +> fails on `SvcEngageCaseUseCase failed: TransitionParticipantRMtoAccepted`; the +> four Invariant Harness rows fail downstream of it on +> `test_invariant_5_expected_event_types_present[engage_case]`, because the 422 +> aborts the trigger before `GuardedCommitCaseLedgerEntryBT` can record the +> entry that #2266 made universally required. Keep them in one row set: routing +> them to separate issues is what left #2216 and #2121 as stale pointers here +> after they were closed. --- diff --git a/notes/received-status-authorization.md b/notes/received-status-authorization.md index 270b5e078..8fd794656 100644 --- a/notes/received-status-authorization.md +++ b/notes/received-status-authorization.md @@ -66,9 +66,9 @@ not wired into the CaseActor's received-side pipeline. ```text AddParticipantStatusBT (Sequence) ├─ VerifySenderIsParticipantNode ← unchanged -├─ CheckParticipantRMNotClosedNode ← unchanged +├─ FilterParticipantStatusDimensionsNode ← per-dimension adjudication (RSH-05) ├─ GuardedCommitOrSkip ← unchanged (CLP-10-006) -├─ AppendParticipantStatusNode ← records "X said FOO" (unchanged) +├─ AppendParticipantStatusNode ← records the accepted portion ├─ StatusUpdateGuard (Fallback) ← NEW │ ├─ CheckIsCaseOwnerNode ← hard bypass: CASE_OWNER = gospel │ └─ CaseOwnerApprovesStatusUpdate ← Evaluator call-out (AlwaysSucceed) @@ -76,6 +76,84 @@ AddParticipantStatusBT (Sequence) └─ AutoCloseIfCaseManager ← unchanged ``` +### Per-dimension partial accept (RSH-05, ADR-0061) + +`FilterParticipantStatusDimensionsNode` replaced the former +`CheckParticipantRMNotClosedNode` guard. The old guard — and +`ValidateRMTransitionNode` inside the append subtree — refused a whole +`ParticipantStatus` snapshot when its `rm` dimension was unacceptable, which +discarded the accepted `vfd`/`pxa` values *and* aborted this Sequence before +the Seam 1 emit, silently skipping embargo teardown (ISSUE-2235). + +The guard now adjudicates `rm`, `vfd` and `pxa` independently and publishes a +*filtered* `ParticipantStatus` in which each refused dimension carries the +participant's current value forward. It is read-only with respect to the +DataLayer (CLP-10-006), so it can run before `GuardedCommitOrSkip` and the +canonical entry snapshots the accepted portion rather than the raw claim. + +`em` is deliberately **not** adjudicated here — embargo state belongs to Seam 2 +(ISSUE-2256). + +Two blackboard keys carry the handoff. Both are written on *every* tick (with +`None` when nothing was filtered) and matched by object ID on read, because the +py_trees blackboard is process-global and `BTBridge.execute_with_setup` restores +only `datalayer` and `trigger_activity_factory` between runs: + +| Key | Producer | Consumers | +|---|---|---| +| `append_status_dimension_filter` | `FilterParticipantStatusDimensionsNode` | `ResolveAndPersistStatusObjectNode`, `ValidateRMTransitionNode` | +| `ledger_payload_object_override` | `FilterParticipantStatusDimensionsNode` | `CommitCaseLedgerEntryNode` | + +`ledger_payload_object_override` is defined in +`vultron/core/behaviors/case/nodes/lifecycle.py` next to its consumer and is +deliberately generic (`{"object_id", "fields"}`): any receive tree may patch the +`object` entry of the ledger payload snapshot, and the other receive trees are +unaffected because the override is opt-in and ID-matched. + +It carries a **field patch, not a replacement object** (RSH-05-009). The +snapshot's `object` is the sender's wire-shaped `ParticipantStatus` — flat +`rmState`/`vfdState`, nested `caseStatus`, plus `@context`, `emConsentState` and +`cvdRole` — and every replica plus the case-ledger invariant harness read it in +that shape. A guard in `vultron.core` cannot rebuild that object: core has zero +`from vultron.wire` imports (ADR-0009, ADR-0017), so dumping the core model +would emit nested `rm`/`vfd` dimension objects and drop every field the guard +never adjudicated. Naming only the adjudicated fields, keyed by wire alias, and +merging them onto the existing snapshot makes shape preservation structural +rather than something the guard has to remember: + +```python +{"object_id": status_id, "fields": {"rmState": "VALID", "vfdState": "VFd", + "caseStatus": {"pxaState": "Pxa", ...}}} +``` + +`CommitCaseLedgerEntryNode._resolve_payload_object_override` merges one level +deep, so patching `caseStatus.pxaState` keeps that nested object's own `id`; it +leaves a `caseStatus` that is still a bare reference string alone, and it drops +the stale snake_case twin of any patched alias so a consumer preferring +`rm_state` cannot read the value the receiver just refused. + +`ValidateRMTransitionNode` keeps its all-or-nothing RM semantics when the +append subtree is used standalone. It only relaxes when the blackboard says +`rm` was refused upstream and carried forward — a narrower change than +reordering its terminal-`CLOSED` and equality checks would have been. + +Two things the guard tracks that are easy to conflate: + +- **An omitted `caseStatus` is not a refusal.** A status that says nothing about + `pxa`/`em` has the receiver's own `case_status` carried forward; persisting the + assertion verbatim would blank both dimensions, which is silent data loss, not + adjudication. So the guard returns two different sets: `refused` names the + dimensions whose asserted value was rejected, while the `model_copy` update + also covers dimensions nobody asserted. If carrying `case_status` forward is + the *only* thing the update does, nothing new was learned and the status is + refused in full (RSH-05-005). +- **A blocked dimension is not always a rewritten one.** `RM.CLOSED` restated by + a participant already at `RM.CLOSED` is refused by the terminal-state rule, but + the recorded value matches the assertion, so nothing was discarded. The + operator-facing WARNING distinguishes `rewrote dimension(s) …` from `blocked + dimension(s) … with no change to the asserted value`; calling the latter a + refusal would misdescribe the audit trail. + ### CASE_OWNER gospel-bypass rationale CASE_OWNER is the human decision-maker for the case. Their reported status diff --git a/notes/sync-ledger-replication.md b/notes/sync-ledger-replication.md index 02dfa21e3..b16242d2d 100644 --- a/notes/sync-ledger-replication.md +++ b/notes/sync-ledger-replication.md @@ -222,7 +222,26 @@ invariants under normal operation and partial failure: 3. **Idempotent replay**: Reprocessing any log prefix (including duplicates) MUST NOT change the resulting state. 4. **Monotonic visibility**: Participants MUST NOT regress their acknowledged - log position. + log position. This extends to projected protocol state: + `ApplyParticipantStatusFromLedgerNode` will not let an entry move a + replica's RM state backwards on the progress scale, even though the Case + Actor is authoritative for *which* transition happened. A replayed, + reordered, or divergent entry would otherwise un-see progress the replica + has already observed. The local value is carried forward for `rm` only; + every other dimension is applied as the entry describes it, and lateral + moves at the same rank (`VALID` ↔ `INVALID`) are re-adjudication rather than + regression (RSH-05-007, ADR-0061). + + The ratcheted status is saved to the DataLayer **unconditionally**. The node + appends the object it reads *back* from the DataLayer — a wire-typed instance + is required, because appending the core model to a + `list[WireParticipantStatus]` makes Pydantic serialize it with the declared + element type's defaults. So the ratchet only takes effect if the ratcheted + copy is what got written. A status object can already be stored locally + without being on the participant (an out-of-order `Announce` of the object + itself, or a replayed entry), and skipping the save in that case appends the + un-ratcheted status while the ratchet's own warning claims the local value was + carried forward. 5. **Reject-on-divergence**: Entries that do not extend the current hash chain MUST be rejected and MUST trigger resynchronization. diff --git a/plan/incoming/learnings/20260812-blackboard-is-process-global-across-bt-runs.md b/plan/incoming/learnings/20260812-blackboard-is-process-global-across-bt-runs.md new file mode 100644 index 000000000..5a59152f4 --- /dev/null +++ b/plan/incoming/learnings/20260812-blackboard-is-process-global-across-bt-runs.md @@ -0,0 +1,33 @@ +--- +title: "py_trees blackboard is process-global and survives BT runs; every guard→effect handoff must be written per tick and ID-matched" +type: learning +timestamp: "2026-08-12T00:00:00Z" +source: ISSUE-2235 +signal: concern +--- + +The ISSUE-2235 fix needed a read-only precondition guard +(`FilterParticipantStatusDimensionsNode`) to hand a filtered `ParticipantStatus` +to two downstream consumers in the same tree. The only available channel is the +py_trees blackboard, and it is a **process-global singleton**. `BTBridge.execute_with_setup` +restores exactly two keys between runs — `datalayer` and `trigger_activity_factory` +— so every other key written by a previous BT execution is still visible to the +next one, in the same process. + +Consequences observed while implementing: + +- A guard that writes its key *only when it has something to say* leaks that + value into the next activity's tree, where a consumer reads a stale payload + that describes a different object entirely. The fix is to write the key on + **every** tick, with `None` when there is nothing to publish. +- Writing per tick is not sufficient on its own: two activities in the same + process can carry different object IDs, so the payload must also record which + object it applies to and consumers must **ID-match** before honouring it. + Both new keys (`append_status_dimension_filter`, + `ledger_payload_object_override`) carry the ID and are matched on read. + +This is not specific to this fix. Any future guard→effect blackboard handoff has +the same hazard, and the failure mode is silent cross-activity contamination +that unit tests running one tree per process will not reproduce. Candidates for +a systemic fix: have `BTBridge` clear (or namespace per-execution) all keys it +did not explicitly seed, rather than restoring a hardcoded pair. diff --git a/plan/incoming/learnings/20260812-datalayer-readback-comment-contradicts-adr-0034.md b/plan/incoming/learnings/20260812-datalayer-readback-comment-contradicts-adr-0034.md new file mode 100644 index 000000000..96f64cc61 --- /dev/null +++ b/plan/incoming/learnings/20260812-datalayer-readback-comment-contradicts-adr-0034.md @@ -0,0 +1,35 @@ +--- +title: "Comments on the DataLayer read-back claim it returns wire-format objects; ADR-0034 says core. The read-back may be vestigial" +type: learning +timestamp: "2026-08-12T00:00:00Z" +source: ISSUE-2235 +signal: concern +--- + +Two places on the participant-status paths save a `ParticipantStatus` and then +immediately read it back from the DataLayer before appending it to +`participant.participant_statuses`, with a comment explaining that the read-back +is needed "to obtain the vocabulary-typed (wire-format) version" — the claim +being that appending a core-model instance to a list declared as +`list[WireParticipantStatus]` makes Pydantic serialize defaults instead of actual +values. The comment now lives in +`vultron/core/behaviors/sync/nodes/participant_status_effect.py`; an equivalent +one was in `status/nodes/append.py`. + +ADR-0034 says the DataLayer port returns **core** domain objects, and that is +what `SqliteDataLayer.read()` was observed to do while diagnosing ISSUE-2235: a +read-back `ParticipantStatus` has `.rm` / `.vfd` dimension objects, not the flat +`.rm_state` / `.vfd_state` of the wire model. So the stated reason for the +read-back cannot be right as written. + +What is not established is whether the read-back is therefore *unnecessary*. It +may still be doing something real (normalizing through the vocabulary registry, +or catching a save failure), or it may be a leftover from before ADR-0034 that +now costs an extra query per apply. I did not change it — ISSUE-2235 had no +reason to touch it and the round-trip is load-bearing for the tests as they +stand. + +Cost paid this session: the misleading comments sent the initial diagnosis toward +a wire/core serialization theory before the actual all-or-nothing control-flow +defect was found. Someone should determine which of the two — the comment or the +read-back — is the thing that is wrong, and delete it. diff --git a/plan/incoming/learnings/20260812-generic-ledger-payload-object-override-seam.md b/plan/incoming/learnings/20260812-generic-ledger-payload-object-override-seam.md new file mode 100644 index 000000000..41b51c72a --- /dev/null +++ b/plan/incoming/learnings/20260812-generic-ledger-payload-object-override-seam.md @@ -0,0 +1,36 @@ +--- +title: "Introduced a generic ledger payload_snapshot override seam rather than a status-specific one" +type: learning +timestamp: "2026-08-12T00:00:00Z" +source: ISSUE-2235 +signal: design-question +--- + +ISSUE-2235 required the canonical `CaseLedgerEntry` to snapshot the *accepted* +portion of an inbound `ParticipantStatus`, not the raw assertion. `CommitCaseLedgerEntryNode` +builds `payload_snapshot` from the activity, so something had to let a preceding +guard substitute the `object` entry. Two shapes were available: + +1. A status-specific hook on `CommitCaseLedgerEntryNode` (e.g. read + `append_status_dimension_filter` directly). +2. A **generic** override key, `ledger_payload_object_override`, carrying + `{"object_id", "object"}`, which any receive tree may publish. + +Option 2 was chosen and the key is defined in +`vultron/core/behaviors/case/nodes/lifecycle.py` next to its consumer, not next +to its (currently sole) producer — so the import direction stays status → case, +matching the pre-existing edge, and no cycle is introduced. + +This was a decision beyond what the issue asked for. Rationale: the same need +recurs for every seam that adjudicates before committing (Seam 2 `em`, +ISSUE-2256, is the immediate next case), and a per-concern hook on the commit +node would accumulate one branch per concern inside the single writer of the +hash chain. + +The risk it accepts: the commit node — the one place that must be trustworthy, +since its output is hash-chained and replicated to every participant — now has an +opt-in path by which an upstream node can rewrite what gets committed. It is +mitigated by being ID-matched (the override is ignored unless it names the object +being committed) and by logging the substitution at INFO. Anything wired to +produce this key is effectively asserting canonical content and should be +reviewed with that in mind. diff --git a/plan/incoming/learnings/20260812-multi-dimension-status-adjudication-unspecified.md b/plan/incoming/learnings/20260812-multi-dimension-status-adjudication-unspecified.md new file mode 100644 index 000000000..b8209df6f --- /dev/null +++ b/plan/incoming/learnings/20260812-multi-dimension-status-adjudication-unspecified.md @@ -0,0 +1,32 @@ +--- +title: "RSH-01 specified who may update status, never what happens to the other dimensions when one is refused" +type: learning +timestamp: "2026-08-12T00:00:00Z" +source: ISSUE-2235 +signal: spec-gap +--- + +`specs/received-status-handling.yaml` RSH-01-001..004 fully specified the +*authorization* half of Seam 1 — who may assert a `ParticipantStatus`, and that +the assertion must be adjudicated before the canonical write. It said nothing +about the fact that a `ParticipantStatus` is a snapshot of **five independent +state machines** (`rm`, `vfd`, `em`, `pxa`, `consent`), and therefore nothing +about what happens to the other four when one of them is unacceptable. + +With no requirement to point at, the implementation defaulted to the shape the +BT gives you for free: a condition node returning FAILURE, which discards the +whole snapshot and aborts the enclosing Sequence. That silently dropped accepted +`vfd`/`pxa` values *and* skipped the Seam 1 → Seam 2 emit, killing embargo +teardown. Nothing in the spec was violated, because nothing in the spec covered +it. + +Filled by RSH-05-001..008 and ADR-0061. The generalizable lesson: whenever a +spec group governs a **composite** object, it needs an explicit statement of +whether the object is adjudicated as a unit or per component. "Validate X" is +ambiguous for any X that is a tuple of independent values, and the BT node +vocabulary biases the ambiguity toward all-or-nothing. + +Other composites in the codebase worth auditing for the same silent +all-or-nothing default: `CaseStatus` (`em` + `pxa`) at Seam 2 — already tracked +as ISSUE-2256 — and `VulnerabilityCase` field updates on +`Announce(VulnerabilityCase)`. diff --git a/plan/incoming/learnings/20260812-nodes-modules-sitting-at-the-500-line-cap.md b/plan/incoming/learnings/20260812-nodes-modules-sitting-at-the-500-line-cap.md new file mode 100644 index 000000000..7e38b5ca8 --- /dev/null +++ b/plan/incoming/learnings/20260812-nodes-modules-sitting-at-the-500-line-cap.md @@ -0,0 +1,33 @@ +--- +title: "Two nodes/ modules sat at 499 and 495 lines against BTND-07-004's 500-line cap, so any change to them forces a decomposition" +type: learning +timestamp: "2026-08-12T00:00:00Z" +source: ISSUE-2235 +signal: concern +--- + +BTND-07-004 caps modules under `nodes/` at 500 lines, enforced by +`test/core/behaviors/test_btnd07_structure.py::test_leaf_module_line_count`. At +the start of this fix: + +- `vultron/core/behaviors/status/nodes/append.py` — 499 lines +- `vultron/core/behaviors/sync/nodes/effects.py` — 495 lines + +Both are exactly the modules ISSUE-2235 had to touch. Adding a docstring +paragraph and a ~25-line method was enough to blow the cap in both, so the bug +fix had to carry an unrelated file-splitting change (`rm_validation.py`, +`participant_status_effect.py`) to get a green suite. That inflates the diff a +reviewer has to read for a behavioural fix, and it happened at the least +convenient moment — after the fix was verified, in the final lint/test loop. + +The cap itself is doing its job; the problem is that there is no warning band. A +module at 495 lines is a decomposition that was deferred, and the next person to +touch it pays for it. Worth considering: a second, softer assertion (or a CI +annotation) at ~90% of the limit so modules get split by whoever is already in +context, not by whoever arrives next with an unrelated change. + +Note also that decomposition is not free of churn beyond the module: both splits +required updating package `__init__.py` re-exports, module docstrings enumerating +the moved classes, and test files that imported the class directly rather than +through the package. `test/core/behaviors/sync/nodes/test_effects.py` turned out +to test only the one class being moved and was renamed to match. diff --git a/plan/incoming/learnings/20260812-pytest-5s-timeout-aborts-whole-suite-under-load.md b/plan/incoming/learnings/20260812-pytest-5s-timeout-aborts-whole-suite-under-load.md new file mode 100644 index 000000000..f01a8e361 --- /dev/null +++ b/plan/incoming/learnings/20260812-pytest-5s-timeout-aborts-whole-suite-under-load.md @@ -0,0 +1,40 @@ +--- +title: "pytest timeout=5 with timeout_method=thread aborts the entire suite under CPU contention, with no failure summary" +type: learning +timestamp: "2026-08-12T00:00:00Z" +source: ISSUE-2235 +signal: tooling-issue +--- + +`pyproject.toml` sets `timeout = 5` and `timeout_method = "thread"`. The thread +method cannot cancel a single test, so when any test exceeds 5 seconds +pytest-timeout dumps every thread's stack and **kills the process**. The run ends +partway through (observed at 6%, 38% and 71% on three separate runs) with a +faulthandler traceback, no `short test summary info`, and no pass/fail counts. + +The margin is thin, which is why it fires so readily. `--durations=10` on a +quiet run puts the slowest tests at **3.84s, 3.01s and 2.96s** against the 5s +cap — `test/metadata/specs/test_real_specs.py::test_real_specs_lint_no_hard_errors` +and its neighbours, plus `test/architecture/test_activity_factory_imports.py` +at 2.91s. Any of those needs only ~30% CPU contention to cross the line and take +the whole run down with it. + +This fires on load, not on defect. It was triggered three times during this +session by ordinary background work in the devcontainer — `uv run pyright` +running concurrently, and the graphify post-checkout rebuild that +`freshen-branch.sh` kicks off three times as it switches branches. Each abort +named a different, unrelated test +(`test/architecture/test_activity_factory_imports.py`, a `starlette.testclient` +HTTP test, a SQLite `cursor.execute`). Run with the machine quiet, or with +`--timeout=60`, the same suite passes with zero failures and zero timeouts. + +Two costs: the output looks like a hard failure of whatever test happened to be +running, which invites debugging the wrong thing; and there is no signal +distinguishing "too slow" from "hung". Do not run other heavy tooling +concurrently with the suite in this container, and read an abrupt end-of-log +faulthandler dump as contention until proven otherwise. + +Possible systemic fixes: raise the default `timeout`, mark the handful of +genuinely slow tests with `@pytest.mark.timeout(N)` and lower the global value, +or switch to `timeout_method = "signal"` so a single test fails instead of the +process dying. diff --git a/specs/received-status-handling.yaml b/specs/received-status-handling.yaml index fb4d750bc..e93576c82 100644 --- a/specs/received-status-handling.yaml +++ b/specs/received-status-handling.yaml @@ -209,3 +209,150 @@ groups: as a parallel refactor target: once `EmitCaseStatusUpdateNode` is in place, `EmitAddCaseStatusToSelfNode` should be refactored to use direct ledger writes as well. That refactor is blocked-by the impl issue. + +- id: RSH-05 + title: Per-Dimension Partial Accept (liberal accept) + description: > + An inbound `Add(ParticipantStatus, CaseParticipant)` carries a snapshot of + several independent state machines (`rm`, `vfd`, `em`, `pxa`, `consent`). + These requirements govern adjudicating each dimension on its own rather + than accepting or discarding the snapshot as a unit. + Derived from ISSUE-2235 and epic ISSUE-2229 (Postel's maxim / liberal + accept). + specs: + - id: RSH-05-001 + priority: MUST + kind: protocol + statement: > + A CaseActor receiving an `Add(ParticipantStatus, CaseParticipant)` MUST + adjudicate each state dimension of the reported status independently. An + unacceptable value in one dimension MUST NOT cause the values reported in + the other dimensions to be discarded. + rationale: > + RM, VFD, EM, PXA and PEC are independent state machines. A value that is + unacceptable in one dimension carries no information about the others, so + discarding the whole snapshot destroys state the receiver has no grounds + to refuse. Liberal accept (ISSUE-2229): accept everything acceptable. + notes: > + Implemented by `FilterParticipantStatusDimensionsNode` in + `vultron/core/behaviors/status/nodes/dimension_filter.py`, wired as a + precondition guard of `add_participant_status_tree`. It supersedes + `CheckParticipantRMNotClosedNode`, whose all-or-nothing RM refusal was + the defect reported in ISSUE-2235. + - id: RSH-05-002 + priority: MUST + kind: protocol + statement: > + When a dimension is refused, the receiver MUST record the participant's + current value for that dimension alongside the accepted values from the + other dimensions, as a single `ParticipantStatus`. + rationale: > + The recorded status must remain a coherent snapshot of the receiver's + view of the participant. Omitting the refused dimension would leave it + undefined; recording the asserted value would adopt a claim the receiver + just refused. + - id: RSH-05-003 + priority: MUST + kind: protocol + statement: > + A refused dimension MUST NOT prevent the Seam 1 emit (RSH-01-003) from + running. The self-addressed `Add(CaseStatus)` MUST still be emitted + whenever any dimension of the update was accepted. + rationale: > + Seam 2 owns embargo teardown (RSH-01-004). Aborting the Seam 1 sequence + on a refused dimension silently skipped teardown — the concrete failure + reported in ISSUE-2235. + - id: RSH-05-004 + priority: MUST + kind: protocol + statement: > + The canonical `CaseLedgerEntry` committed for a partially accepted status + MUST snapshot the accepted portion, not the raw assertion. + rationale: > + The canonical projection is replicated to every participant and bound + into the hash chain. Snapshotting the raw assertion would fan out a value + the receiver refused, and each replica would apply it. Recording the + accepted portion is also what makes the refusal visible: the canonical + entry differs from what the sender asserted. + notes: > + Implemented via the `ledger_payload_object_override` blackboard contract + read by `CommitCaseLedgerEntryNode`. The guard runs before + `GuardedCommit` per CLP-10-006, so the adjudication is available at + commit time. The guard is read-only with respect to the DataLayer. + + The override is a *patch* — `{"object_id": ..., "fields": {...}}` keyed by + wire alias — merged onto the snapshot's existing `object`, not a + replacement object. See RSH-05-009 for why. + - id: RSH-05-005 + priority: MUST + kind: protocol + statement: > + A status update in which every refused dimension leaves the resulting + snapshot indistinguishable from the participant's current state MUST be + refused in full: nothing is appended and no canonical ledger entry is + committed. + rationale: > + Such an assertion carries no acceptable information. Appending it would + grow the participant's status history and the hash chain without + recording any state change. + - id: RSH-05-006 + priority: MUST + kind: protocol + statement: > + `RM.CLOSED` MUST be treated as terminal for the `rm` dimension only. A + participant at `RM.CLOSED` MUST still have accepted values in other + dimensions recorded. + rationale: > + A vendor that has closed its report management workflow may still deploy + a fix (VFD) or observe public disclosure (PXA). Freezing the whole + snapshot at RM closure discards real protocol state. + - id: RSH-05-007 + priority: MUST + kind: protocol + statement: > + A participant applying an `Announce(CaseLedgerEntry)` for an + `add_participant_status_to_participant` event MUST NOT let it move a + replica's RM state backwards on the RM progress scale. The local value + MUST be carried forward for that dimension; all other dimensions are + applied as the entry describes them. + rationale: > + Monotonic visibility (see notes/sync-ledger-replication.md). The Case + Actor is authoritative for which transition happened, but a replayed, + reordered, or divergent entry would otherwise un-see progress the replica + has already observed. Lateral moves at the same rank + (`VALID` <-> `INVALID`) are re-adjudication, not regression, and are + applied unchanged. + - id: RSH-05-008 + priority: SHOULD + kind: protocol + statement: > + The `em` dimension SHOULD NOT be adjudicated at Seam 1; embargo state + belongs to Seam 2 (`add_case_status_tree`). + rationale: > + EM is per-case rather than per-participant and its authorization seam is + the SideEffectsGuard (RSH-02-001). Adjudicating it in the Seam 1 guard + would duplicate — and could contradict — Seam 2's decision. + notes: > + Seam 2 EM adjudication is tracked in ISSUE-2256. + - id: RSH-05-009 + priority: MUST + kind: protocol + statement: > + Adjudicating a received status MUST NOT change the shape of the + `payload_snapshot.object` it records: the entry MUST carry the same wire + fields an unadjudicated entry for the same activity would carry, with only + the adjudicated dimension values rewritten. + rationale: > + Every replica and the case-ledger invariant harness read the snapshot's + flat wire aliases (`rmState`, `vfdState`, `emConsentState`, `cvdRole`) and + its nested `caseStatus` (CLP-07-001, CM-18-006). A whole-object + replacement built in `vultron.core` would emit core dimension objects + (`rm: {state}`) and silently drop every field the guard never adjudicated, + because core MUST NOT import `vultron.wire` to convert (ADR-0009, + ADR-0017). Publishing a field patch instead makes shape preservation + structural rather than a property the guard has to remember to maintain. + notes: > + The merge is one level deep, so a patched `caseStatus.pxaState` keeps the + nested object's own `id` and remaining fields. A patched alias also drops + any stale snake_case twin of the same field, so a consumer that prefers + the snake_case spelling cannot read the value the receiver refused. diff --git a/test/core/behaviors/status/nodes/test_append.py b/test/core/behaviors/status/nodes/test_append.py index 38431a173..508a0dce9 100644 --- a/test/core/behaviors/status/nodes/test_append.py +++ b/test/core/behaviors/status/nodes/test_append.py @@ -16,9 +16,9 @@ """Unit tests for append-participant-status leaf nodes. Tests SkipIfIdempotentNode, LoadParticipantNode, -CheckStatusNotAlreadyAppendedNode, ResolveAndPersistStatusObjectNode, -ValidateRMTransitionNode, and AppendStatusAndSaveParticipantNode -from nodes.append. +CheckStatusNotAlreadyAppendedNode, ResolveAndPersistStatusObjectNode and +AppendStatusAndSaveParticipantNode from ``nodes.append``, plus +ValidateRMTransitionNode from ``nodes.rm_validation``. Per DEMOMA-07-003 step 2. """ @@ -35,6 +35,8 @@ LoadParticipantNode, ResolveAndPersistStatusObjectNode, SkipIfIdempotentNode, +) +from vultron.core.behaviors.status.nodes.rm_validation import ( ValidateRMTransitionNode, ) from vultron.enums.roles import CVDRole diff --git a/test/core/behaviors/status/test_partial_accept_participant_status.py b/test/core/behaviors/status/test_partial_accept_participant_status.py new file mode 100644 index 000000000..16ac01651 --- /dev/null +++ b/test/core/behaviors/status/test_partial_accept_participant_status.py @@ -0,0 +1,961 @@ +#!/usr/bin/env python + +# Copyright (c) 2026 Carnegie Mellon University and Contributors. +# - see Contributors.md for a full list of Contributors +# - see ContributionInstructions.md for information on how you can Contribute to this project +# Vultron Multiparty Coordinated Vulnerability Disclosure Protocol Prototype is +# licensed under a MIT (SEI)-style license, please see LICENSE.md distributed +# with this Software or contact permission@sei.cmu.edu for full terms. +# Created, in part, with funding and support from the United States Government +# (see Acknowledgments file). This program may include and/or can make use of +# certain third party source code, object code, documentation and other files +# ("Third Party Software"). See LICENSE.md for more details. +# Carnegie Mellon®, CERT® and CERT Coordination Center® are registered in the +# U.S. Patent and Trademark Office by Carnegie Mellon University + +"""Regression tests for ISSUE-2235 — per-dimension partial accept (RSH-05). + +#2235: "Rejected status updates are dropped silently and all-or-nothing +(violates liberal-accept)." + +An inbound ``Add(ParticipantStatus, CaseParticipant)`` carries a *snapshot* of +several independent state machines (``rm``, ``vfd``, ``em``, ``pxa``, +``consent``). Before this fix, a single refused dimension — a regressive +``rm`` or a status for a participant already at terminal ``RM.CLOSED`` — +caused the receiving Case Actor to discard the entire snapshot and abort the +``AddParticipantStatusBT`` Sequence, which also killed the Seam 1 → Seam 2 +emit (``EmitAddCaseStatusToSelfNode``) and therefore embargo teardown +(ADR-0046, RSH-01-003). + +The fix accepts each dimension independently: refused dimensions carry +forward the participant's current value, accepted dimensions are recorded, +and the canonical ledger entry snapshots the *accepted* portion rather than +the raw assertion. The refusal is visible in the canonical ledger (the +accepted portion differs from what was asserted); no new wire message is +emitted. + +Per specs/received-status-handling.yaml RSH-05. +""" + +from typing import Any, cast + +import py_trees +import pytest +from py_trees.common import Status + +from vultron.adapters.driven.datalayer_sqlite import SqliteDataLayer +from vultron.adapters.driven.trigger_activity_adapter import ( + TriggerActivityAdapter, +) +from vultron.core.behaviors.bridge import BTBridge +from vultron.core.behaviors.case.nodes.lifecycle import ( + BB_LEDGER_PAYLOAD_OBJECT_OVERRIDE, + _merge_snapshot_object_fields, +) +from vultron.core.behaviors.status.add_participant_status_tree import ( + add_participant_status_tree, +) +from vultron.core.behaviors.status.nodes.dimension_filter import ( + BB_DIMENSION_FILTER, + FilterParticipantStatusDimensionsNode, + resolve_dimension_filter, +) +from vultron.core.behaviors.sync.nodes.participant_status_effect import ( + ApplyParticipantStatusFromLedgerNode, +) +from vultron.core.models.case_ledger import HashChainLedgerRecord +from vultron.core.models.case_ledger_entry import VultronCaseLedgerEntry +from vultron.core.models.case_participant import CaseParticipant +from vultron.core.behaviors.sync.nodes.chain import _to_persistable_entry +from vultron.core.models.events.sync import AnnounceLogEntryReceivedEvent +from vultron.core.states.cs import CS_pxa, CS_vfd +from vultron.core.states.em import EM +from vultron.core.states.participant_embargo_consent import PEC +from vultron.core.states.rm import RM +from vultron.enums.roles import CVDRole +from vultron.semantic_registry import extract_event +from vultron.wire.as2.factories import ( + add_status_to_participant_activity, + announce_log_entry_activity, +) +from vultron.wire.as2.vocab.objects.case_ledger_entry import ( + as_CaseLedgerEntry as WireCaseLedgerEntry, +) +from vultron.wire.as2.vocab.objects.case_participant import as_CaseParticipant +from vultron.wire.as2.vocab.objects.case_status import ( + as_CaseStatus, + as_ParticipantStatus, +) +from vultron.wire.as2.vocab.objects.vulnerability_case import ( + as_VulnerabilityCase, +) + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +ACTOR_ID = "https://example.org/actors/vendor" +CASE_MANAGER_ID = "https://example.org/actors/case-actor" +CASE_ID = "https://example.org/cases/case-2235" +PARTICIPANT_ID = f"{CASE_ID}/participants/vendor" +CM_PARTICIPANT_ID = f"{CASE_ID}/participants/case-actor" +CURRENT_STATUS_ID = f"{PARTICIPANT_ID}/statuses/current" +ASSERTED_STATUS_ID = f"{PARTICIPANT_ID}/statuses/asserted" +SECOND_STATUS_ID = f"{PARTICIPANT_ID}/statuses/asserted-2" + +_ZERO_HASH = "0" * 64 + + +# --------------------------------------------------------------------------- +# Shape-agnostic accessors +# +# ``SqliteDataLayer.read`` returns *core* models (``rm``/``vfd`` dimension +# objects) while wire objects and wire-shaped ledger snapshots use the flat +# ``rmState``/``vfdState`` form. These readers accept either and normalize to +# the enum *member name* (``"VALID"``, ``"VFd"``, ``"Pxa"``) — which is also +# what both serializations carry — so the assertions describe protocol state, +# not serialization shape. Comparing enum members directly would not work for +# ``CS_vfd``/``CS_pxa``, whose ``.value`` is a ``NamedTuple`` rather than the +# string that appears on the wire. +# --------------------------------------------------------------------------- + + +def _state_name(value: Any) -> str | None: + """Normalize an enum member or wire string to the member name.""" + if value is None: + return None + return getattr(value, "name", None) or str(value) + + +def _dim_state(obj: Any, core_field: str, flat_field: str) -> str | None: + """Return a dimension's state name from a core or wire object/dict.""" + if isinstance(obj, dict): + nested = obj.get(core_field) + if isinstance(nested, dict): + return _state_name(nested.get("state")) + camel = flat_field[0] + flat_field.title().replace("_", "")[1:] + return _state_name(obj.get(flat_field) or obj.get(camel)) + nested = getattr(obj, core_field, None) + if nested is not None and hasattr(nested, "state"): + return _state_name(nested.state) + return _state_name(getattr(obj, flat_field, None)) + + +def _rm_of(obj: Any) -> str | None: + return _dim_state(obj, "rm", "rm_state") + + +def _vfd_of(obj: Any) -> str | None: + return _dim_state(obj, "vfd", "vfd_state") + + +def _case_status_of(obj: Any) -> Any: + if isinstance(obj, dict): + return obj.get("case_status") or obj.get("caseStatus") + return getattr(obj, "case_status", None) + + +def _pxa_of(obj: Any) -> str | None: + cs = _case_status_of(obj) + return None if cs is None else _dim_state(cs, "pxa", "pxa_state") + + +def _em_of(obj: Any) -> str | None: + cs = _case_status_of(obj) + return None if cs is None else _dim_state(cs, "em", "em_state") + + +def _latest_status(dl: SqliteDataLayer, participant_id: str) -> Any: + participant = cast(CaseParticipant, dl.read(participant_id)) + assert participant is not None + assert participant.participant_statuses + return participant.participant_statuses[-1] + + +def _status_ids(dl: SqliteDataLayer, participant_id: str) -> list[str]: + participant = cast(CaseParticipant, dl.read(participant_id)) + assert participant is not None + return [ + str(getattr(s, "id_", s)) for s in participant.participant_statuses + ] + + +def _ledger_entries(dl: SqliteDataLayer) -> list[VultronCaseLedgerEntry]: + entries = [ + cast(VultronCaseLedgerEntry, obj) + for obj in dl.list_objects("CaseLedgerEntry") + if isinstance(obj, VultronCaseLedgerEntry) + and cast(VultronCaseLedgerEntry, obj).case_id == CASE_ID + ] + return sorted(entries, key=lambda e: e.log_index) + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture(autouse=True) +def clear_blackboard(): + py_trees.blackboard.Blackboard.storage.clear() + yield + py_trees.blackboard.Blackboard.storage.clear() + + +@pytest.fixture +def dl(): + return SqliteDataLayer("sqlite:///:memory:") + + +def _current_status( + rm_state: RM, + vfd_state: CS_vfd, + pxa_state: CS_pxa, +) -> as_ParticipantStatus: + """The participant's status *before* the inbound assertion arrives.""" + return as_ParticipantStatus( + id_=CURRENT_STATUS_ID, + context=CASE_ID, + rm_state=rm_state, + vfd_state=vfd_state, + em_consent_state=PEC.SIGNATORY, + case_status=as_CaseStatus( + id_=f"{CURRENT_STATUS_ID}/cs", + context=CASE_ID, + em_state=EM.NONE, + pxa_state=pxa_state, + ), + ) + + +def _asserted_status( + rm_state: RM, + vfd_state: CS_vfd, + pxa_state: CS_pxa | None, + status_id: str = ASSERTED_STATUS_ID, +) -> as_ParticipantStatus: + """The inbound assertion from the sender. + + ``pxa_state=None`` builds a status with **no** ``case_status`` at all — the + normal shape when the sender has nothing to say about the case-level + dimensions, not a malformed message. + """ + case_status = ( + None + if pxa_state is None + else as_CaseStatus( + id_=f"{status_id}/cs", + context=CASE_ID, + em_state=EM.NONE, + pxa_state=pxa_state, + ) + ) + return as_ParticipantStatus( + id_=status_id, + context=CASE_ID, + rm_state=rm_state, + vfd_state=vfd_state, + em_consent_state=PEC.SIGNATORY, + case_status=case_status, + ) + + +def _seed_case( + dl: SqliteDataLayer, + current: as_ParticipantStatus, + asserted: as_ParticipantStatus | None, +) -> None: + """Seed a two-participant case with *current* as the vendor's latest status.""" + vendor = as_CaseParticipant( + id_=PARTICIPANT_ID, + context=CASE_ID, + attributed_to=ACTOR_ID, + case_roles=[CVDRole.CASE_OWNER], + ) + vendor.participant_statuses.append(current) + manager = as_CaseParticipant( + id_=CM_PARTICIPANT_ID, + context=CASE_ID, + attributed_to=CASE_MANAGER_ID, + case_roles=[CVDRole.CASE_MANAGER], + ) + # attributed_to is what seeds the per-case genesis hash (CLP-08-003); + # without it the ledger sits in the pre-genesis bootstrap window and the + # guarded commit cannot anchor a chain. + case = as_VulnerabilityCase( + id_=CASE_ID, + name="Issue 2235 Case", + attributed_to=CASE_MANAGER_ID, + ) + case.add_participant(vendor) + case.add_participant(manager) + + dl.create(case) + dl.create(vendor) + dl.create(manager) + dl.create(current) + if asserted is not None: + dl.create(asserted) + + +def _run_tree( + dl: SqliteDataLayer, + asserted: as_ParticipantStatus, + executing_actor_id: str, + make_payload: Any, +) -> Any: + """Run the full ``add_participant_status_tree`` for *asserted*.""" + activity = add_status_to_participant_activity( + status=asserted, + target=as_CaseParticipant( + id_=PARTICIPANT_ID, context=CASE_ID, attributed_to=ACTOR_ID + ), + actor=ACTOR_ID, + context=as_VulnerabilityCase(id_=CASE_ID, name="Issue 2235 Case"), + ) + event = make_payload(activity) + bridge = BTBridge( + datalayer=dl, trigger_activity=TriggerActivityAdapter(dl) + ) + tree = add_participant_status_tree(request=event, case_id=CASE_ID) + # Production passes the parsed event as ``activity`` (see + # SvcAddParticipantStatusToParticipantReceivedUseCase); the guarded commit + # needs it on the blackboard to build a payload snapshot. + return bridge.execute_with_setup( + tree=tree, actor_id=executing_actor_id, activity=event + ) + + +# --------------------------------------------------------------------------- +# A refused rm must not discard accepted vfd / pxa +# --------------------------------------------------------------------------- + + +class TestRefusedDimensionDoesNotDiscardAcceptedDimensions: + """A regressive ``rm`` must not throw away the rest of the snapshot.""" + + def test_regressive_rm_carried_forward_accepted_vfd_and_pxa_recorded( + self, dl, make_payload + ): + """VALID + rm=RECEIVED (refused) + vfd=VFd + pxa=Pxa (both accepted). + + The status is appended with the participant's current ``rm`` carried + forward and the two forward dimensions applied (RSH-05). + """ + current = _current_status(RM.VALID, CS_vfd.Vfd, CS_pxa.pxa) + asserted = _asserted_status(RM.RECEIVED, CS_vfd.VFd, CS_pxa.Pxa) + _seed_case(dl, current, asserted) + + result = _run_tree(dl, asserted, ACTOR_ID, make_payload) + assert result.status == Status.SUCCESS, ( + "a refused rm dimension must not abort the whole update" + f" (feedback: {result.feedback_message})" + ) + + assert ASSERTED_STATUS_ID in _status_ids(dl, PARTICIPANT_ID) + latest = _latest_status(dl, PARTICIPANT_ID) + assert _rm_of(latest) == RM.VALID.name, "refused rm must carry forward" + assert ( + _vfd_of(latest) == CS_vfd.VFd.name + ), "accepted vfd must be recorded" + assert ( + _pxa_of(latest) == CS_pxa.Pxa.name + ), "accepted pxa must be recorded" + assert ( + _em_of(latest) == EM.NONE.name + ), "em is Seam 2's business (#2256)" + + def test_regressive_rm_still_reaches_seam_2_emit(self, dl, make_payload): + """The Seam 1 → Seam 2 emit must survive a refused dimension. + + This is the concrete failure reported in #2235: aborting the Sequence + at RM validation skipped ``EmitAddCaseStatusToSelfNode``, so embargo + teardown in Seam 2 never ran (RSH-01-003, RSH-01-004). + """ + current = _current_status(RM.VALID, CS_vfd.Vfd, CS_pxa.pxa) + asserted = _asserted_status(RM.RECEIVED, CS_vfd.VFd, CS_pxa.Pxa) + _seed_case(dl, current, asserted) + + result = _run_tree(dl, asserted, ACTOR_ID, make_payload) + assert result.status == Status.SUCCESS + + outbox = dl.outbox_list_for_actor(ACTOR_ID) + assert len(outbox) > 0, ( + "EmitAddCaseStatusToSelfNode must still queue Add(CaseStatus)" + " when one dimension was refused" + ) + + +# --------------------------------------------------------------------------- +# The canonical ledger records the accepted portion +# --------------------------------------------------------------------------- + + +class TestCanonicalLedgerRecordsAcceptedPortion: + """The refusal is made visible by what the canonical ledger records.""" + + def test_ledger_snapshot_carries_accepted_rm_not_asserted_rm( + self, dl, make_payload + ): + """Run as CASE_MANAGER so the guarded commit fires (CLP-10-006). + + The committed ``payload_snapshot['object']`` must describe the + *accepted* status, not the sender's raw assertion — otherwise the + refused value is replicated to every participant. + """ + current = _current_status(RM.VALID, CS_vfd.Vfd, CS_pxa.pxa) + asserted = _asserted_status(RM.RECEIVED, CS_vfd.VFd, CS_pxa.Pxa) + _seed_case(dl, current, asserted) + + result = _run_tree(dl, asserted, CASE_MANAGER_ID, make_payload) + assert result.status == Status.SUCCESS + + entries = _ledger_entries(dl) + assert len(entries) == 1, "exactly one canonical entry expected" + snapshot_object = entries[0].payload_snapshot.get("object") + assert isinstance(snapshot_object, dict), ( + "the ledger snapshot must inline the status object," + f" got {snapshot_object!r}" + ) + assert _rm_of(snapshot_object) == RM.VALID.name, ( + "ledger must record the accepted rm (VALID), not the refused" + f" assertion — got {_rm_of(snapshot_object)!r}" + ) + assert _vfd_of(snapshot_object) == CS_vfd.VFd.name + assert _pxa_of(snapshot_object) == CS_pxa.Pxa.name + + def test_ledger_snapshot_keeps_the_wire_shape_of_an_unfiltered_snapshot( + self, dl, make_payload + ): + """Adjudication must rewrite values, never reshape the snapshot. + + ``payload_snapshot['object']`` is consumed by every replica and by the + invariant harness, which read the flat wire aliases (``rmState``, + ``vfdState``, ``emConsentState``, ``cvdRole``) and the nested + ``caseStatus``. The guard runs in ``vultron.core`` and cannot import + the wire layer to rebuild the object, so it publishes a *patch* over the + sender's already-wire-shaped snapshot. A snapshot built by dumping the + core model instead would carry nested ``rm``/``vfd`` dimension objects + and silently drop every field the guard never adjudicated + (CLP-07-001, CM-18-006, ADR-0009). + """ + current = _current_status(RM.VALID, CS_vfd.Vfd, CS_pxa.pxa) + asserted = _asserted_status(RM.RECEIVED, CS_vfd.VFd, CS_pxa.Pxa) + _seed_case(dl, current, asserted) + + result = _run_tree(dl, asserted, CASE_MANAGER_ID, make_payload) + assert result.status == Status.SUCCESS + + entries = _ledger_entries(dl) + assert len(entries) == 1 + snap = entries[0].payload_snapshot["object"] + assert isinstance(snap, dict) + + # Flat wire aliases, carrying the adjudicated values. + assert snap["rmState"] == RM.VALID.name + assert snap["vfdState"] == CS_vfd.VFd.name + + # Fields the guard never adjudicated survive the patch untouched. + assert ( + snap.get("emConsentState") == PEC.SIGNATORY.name + ), "emConsentState must survive adjudication (fcvcv invariant harness)" + assert "cvdRole" in snap, "cvdRole must survive adjudication" + assert "@context" in snap, "@context must survive adjudication" + assert snap.get("type") == "ParticipantStatus" + + # No core-model shapes, and no stale snake_case twin of a patched field. + assert not isinstance( + snap.get("rm"), dict + ), f"core 'rm' dimension object leaked into the snapshot: {snap!r}" + assert not isinstance( + snap.get("vfd"), dict + ), f"core 'vfd' dimension object leaked into the snapshot: {snap!r}" + assert "rm_state" not in snap + assert "vfd_state" not in snap + + # The nested caseStatus is patched in place, keeping its own identity. + case_status = snap["caseStatus"] + assert isinstance(case_status, dict) + assert case_status["pxaState"] == CS_pxa.Pxa.name + assert case_status["emState"] == EM.NONE.name + assert case_status.get("id") == f"{ASSERTED_STATUS_ID}/cs" + assert "pxa_state" not in case_status + + +# --------------------------------------------------------------------------- +# An omitted case_status asserts nothing — it must not erase pxa/em +# --------------------------------------------------------------------------- + + +class TestOmittedCaseStatusIsNotAnAssertion: + """A status with no ``caseStatus`` says nothing about ``pxa``/``em``. + + Persisting such an assertion verbatim would blank both dimensions on the + receiver, which is a silent data loss rather than an adjudication: the + sender never claimed anything to adjudicate (RSH-05-002). + """ + + def test_omitted_case_status_does_not_erase_pxa_and_em( + self, dl, make_payload + ): + """vfd advances; the receiver's own ``case_status`` carries forward.""" + current = _current_status(RM.VALID, CS_vfd.Vfd, CS_pxa.pXa) + asserted = _asserted_status(RM.VALID, CS_vfd.VFd, None) + assert asserted.case_status is None + _seed_case(dl, current, asserted) + + result = _run_tree(dl, asserted, ACTOR_ID, make_payload) + assert result.status == Status.SUCCESS, ( + "an omitted case_status is not a refusal" + f" (feedback: {result.feedback_message})" + ) + + latest = _latest_status(dl, PARTICIPANT_ID) + assert ( + _vfd_of(latest) == CS_vfd.VFd.name + ), "the vfd advance is accepted" + assert ( + _pxa_of(latest) == CS_pxa.pXa.name + ), "an unasserted pxa must be carried forward, not blanked" + assert ( + _em_of(latest) == EM.NONE.name + ), "an unasserted em must be carried forward, not blanked" + + def test_omitted_case_status_alone_carries_no_new_state( + self, dl, make_payload + ): + """Nothing asserted but the omission → refused in full, no entry. + + Carrying ``case_status`` forward is not new information, so appending + the status would grow the history and the hash chain without recording + a state change (RSH-05-005). Run as the Case Manager so a commit + *would* fire if the guards let it through. + """ + current = _current_status(RM.VALID, CS_vfd.Vfd, CS_pxa.pXa) + asserted = _asserted_status(RM.VALID, CS_vfd.Vfd, None) + _seed_case(dl, current, asserted) + + result = _run_tree(dl, asserted, CASE_MANAGER_ID, make_payload) + assert result.status == Status.FAILURE + + assert ASSERTED_STATUS_ID not in _status_ids(dl, PARTICIPANT_ID) + assert _ledger_entries(dl) == [] + latest = _latest_status(dl, PARTICIPANT_ID) + assert _pxa_of(latest) == CS_pxa.pXa.name + assert _em_of(latest) == EM.NONE.name + + +# --------------------------------------------------------------------------- +# Terminal RM.CLOSED +# --------------------------------------------------------------------------- + + +class TestTerminalClosedParticipant: + """``RM.CLOSED`` freezes ``rm`` only — not the other dimensions.""" + + def test_closed_participant_still_accepts_vfd_advance( + self, dl, make_payload + ): + """A CLOSED vendor deploying its fix must still be recorded. + + ``rm`` stays CLOSED (terminal); ``vfd`` advances Vfd → VFd. + """ + current = _current_status(RM.CLOSED, CS_vfd.Vfd, CS_pxa.pxa) + asserted = _asserted_status(RM.CLOSED, CS_vfd.VFd, CS_pxa.pxa) + _seed_case(dl, current, asserted) + + result = _run_tree(dl, asserted, ACTOR_ID, make_payload) + assert result.status == Status.SUCCESS, ( + "a CLOSED participant may still report vfd/pxa progress" + f" (feedback: {result.feedback_message})" + ) + + latest = _latest_status(dl, PARTICIPANT_ID) + assert _rm_of(latest) == RM.CLOSED.name + assert _vfd_of(latest) == CS_vfd.VFd.name + + def test_wholly_refused_update_is_not_appended_and_commits_no_entry( + self, dl, make_payload + ): + """CLOSED + duplicate CLOSED with no other change → refused outright. + + Nothing is appended and no canonical ledger entry is committed: the + assertion carried no acceptable information. Executed as the Case + Manager so a commit *would* fire if the guards let it through. + """ + current = _current_status(RM.CLOSED, CS_vfd.Vfd, CS_pxa.pxa) + asserted = _asserted_status(RM.CLOSED, CS_vfd.Vfd, CS_pxa.pxa) + _seed_case(dl, current, asserted) + + result = _run_tree(dl, asserted, CASE_MANAGER_ID, make_payload) + assert result.status == Status.FAILURE + + assert ASSERTED_STATUS_ID not in _status_ids(dl, PARTICIPANT_ID) + assert _ledger_entries(dl) == [] + + +# --------------------------------------------------------------------------- +# Ledger-apply path (replica side) +# --------------------------------------------------------------------------- + + +def _status_snapshot_entry( + rm_state: str, vfd_state: str +) -> VultronCaseLedgerEntry: + """A canonical ``add_participant_status_to_participant`` entry.""" + return _to_persistable_entry( + HashChainLedgerRecord( + case_id=CASE_ID, + log_index=0, + object_id="https://example.org/activities/add-status-2235", + event_type="add_participant_status_to_participant", + payload_snapshot={ + "object": { + "id": ASSERTED_STATUS_ID, + "type": "ParticipantStatus", + "context": CASE_ID, + "rmState": rm_state, + "vfdState": vfd_state, + }, + "target": {"id": PARTICIPANT_ID}, + }, + prev_log_hash=_ZERO_HASH, + ) + ) + + +def _announce_event( + entry: VultronCaseLedgerEntry, +) -> AnnounceLogEntryReceivedEvent: + wire_entry = WireCaseLedgerEntry.model_validate( + entry.model_dump(mode="json") + ) + activity = announce_log_entry_activity( + entry=wire_entry, actor=CASE_MANAGER_ID + ) + return cast(AnnounceLogEntryReceivedEvent, extract_event(activity)) + + +class TestLedgerApplyRmRatchet: + """A replicated entry must not regress a replica's derived RM state.""" + + def test_regressive_rm_in_ledger_entry_does_not_regress_replica(self, dl): + """Replica at VALID; entry asserts RECEIVED + vfd=VFd. + + Monotonic visibility (see notes/sync-ledger-replication.md): the + replica keeps ``rm`` at VALID while still applying the accepted + ``vfd`` advance. + """ + current = _current_status(RM.VALID, CS_vfd.Vfd, CS_pxa.pxa) + _seed_case(dl, current, None) + + entry = _status_snapshot_entry(rm_state="RECEIVED", vfd_state="VFd") + event = _announce_event(entry) + + bridge = BTBridge(datalayer=dl) + result = bridge.execute_with_setup( + tree=ApplyParticipantStatusFromLedgerNode( + name="ApplyParticipantStatusFromLedger" + ), + actor_id=ACTOR_ID, + activity=event, + ) + assert result.status == Status.SUCCESS + + latest = _latest_status(dl, PARTICIPANT_ID) + assert ( + _rm_of(latest) == RM.VALID.name + ), "a replicated entry must not regress the replica's rm state" + assert ( + _vfd_of(latest) == CS_vfd.VFd.name + ), "the accepted vfd advance must still be applied" + + def test_ratchet_holds_when_the_status_object_is_already_stored_locally( + self, dl + ): + """The ratchet must survive a status object already in the DataLayer. + + The node appends the object it *reads back* from the DataLayer, so the + ratcheted value only reaches ``participant_statuses`` if the ratcheted + copy is saved. A status object can already be stored locally without + being on the participant — an out-of-order ``Announce`` of the object + itself, or a replayed entry — and skipping the save in that case appends + the un-ratcheted status while the ratchet's own log line claims the + local value was carried forward (RSH-05-007, SYNC-02-002). + """ + current = _current_status(RM.VALID, CS_vfd.Vfd, CS_pxa.pxa) + _seed_case(dl, current, None) + # Present as a stored object, absent from participant_statuses. + dl.create(_asserted_status(RM.RECEIVED, CS_vfd.VFd, CS_pxa.pxa)) + assert ASSERTED_STATUS_ID not in _status_ids(dl, PARTICIPANT_ID) + + entry = _status_snapshot_entry(rm_state="RECEIVED", vfd_state="VFd") + event = _announce_event(entry) + + bridge = BTBridge(datalayer=dl) + result = bridge.execute_with_setup( + tree=ApplyParticipantStatusFromLedgerNode( + name="ApplyParticipantStatusFromLedger" + ), + actor_id=ACTOR_ID, + activity=event, + ) + assert result.status == Status.SUCCESS + + latest = _latest_status(dl, PARTICIPANT_ID) + assert _rm_of(latest) == RM.VALID.name, ( + "the ratcheted rm must be persisted even when the status object" + " was already present in the local DataLayer" + ) + assert _vfd_of(latest) == CS_vfd.VFd.name + + def test_unreadable_local_rm_fails_instead_of_skipping_the_ratchet( + self, dl, monkeypatch + ): + """An unreadable RM floor is a shape mismatch, not "no floor". + + The ratchet needs the replica's current RM to know what a regression + *is*. Reading that floor with a defaulting accessor turned a + non-core-shaped local record into ``None``, which made the ratchet a + no-op and applied the regressing entry unchecked — the #2264 failure + mode, silent because the ratchet only logs when it refuses something. + ARCH-15-001 and ARCH-15-002 require FAILURE (ADR-0062). + """ + current = _current_status(RM.VALID, CS_vfd.Vfd, CS_pxa.pxa) + _seed_case(dl, current, None) + + # A *core* participant (so the node does not skip it as "not found") + # whose latest status is wire-shaped: flat ``rmState``, no ``rm`` + # attribute at all. Pydantic does not validate on list append, which + # is how such a record survives into a replica in the first place. + broken = cast(CaseParticipant, dl.read(PARTICIPANT_ID)) + assert isinstance(broken, CaseParticipant) + broken.participant_statuses[-1] = cast(Any, current) + + real_read = dl.read + monkeypatch.setattr( + dl, + "read", + lambda object_id: ( + broken if object_id == PARTICIPANT_ID else real_read(object_id) + ), + ) + + entry = _status_snapshot_entry(rm_state="RECEIVED", vfd_state="VFd") + event = _announce_event(entry) + + bridge = BTBridge(datalayer=dl) + result = bridge.execute_with_setup( + tree=ApplyParticipantStatusFromLedgerNode( + name="ApplyParticipantStatusFromLedger" + ), + actor_id=ACTOR_ID, + activity=event, + ) + assert ( + result.status == Status.FAILURE + ), "an unreadable RM floor must fail, not silently skip the ratchet" + + assert real_read(ASSERTED_STATUS_ID) is None, ( + "the regressing status must not be persisted when the ratchet" + " cannot be enforced" + ) + assert ASSERTED_STATUS_ID not in [ + str(getattr(s, "id_", s)) for s in broken.participant_statuses + ], "the regressing status must not reach participant_statuses" + + +# --------------------------------------------------------------------------- +# Blackboard hygiene +# +# The py_trees blackboard is process-global and is not cleared between tree +# executions, so every key a node writes is a potential leak into the next run +# (BT-17-003, BT-17-004). The ledger override is the dangerous one: it rewrites +# what gets hash-chained and replicated to every participant. +# --------------------------------------------------------------------------- + + +class TestLedgerOverrideDoesNotLeakBetweenExecutions: + """A stale override must never reach a later commit.""" + + def test_second_execution_does_not_inherit_the_first_overrides( + self, dl, make_payload + ): + """Two runs of the same status ID, no blackboard clear in between. + + Run 1 partially accepts and commits the adjudicated snapshot. Run 2 is + an idempotent re-delivery: the filter adjudicates nothing, so run 2's + receipt entry must record the snapshot exactly as it arrived. Both runs + carry the same ``object_id``, so the commit node's ID match cannot catch + this leak — the filter has to clear the key on its no-op path + (BT-17-003, BT-17-004). + """ + current = _current_status(RM.VALID, CS_vfd.Vfd, CS_pxa.pxa) + asserted = _asserted_status(RM.RECEIVED, CS_vfd.VFd, CS_pxa.Pxa) + _seed_case(dl, current, asserted) + + first = _run_tree(dl, asserted, CASE_MANAGER_ID, make_payload) + assert first.status == Status.SUCCESS + assert ASSERTED_STATUS_ID in _status_ids(dl, PARTICIPANT_ID) + + second = _run_tree(dl, asserted, CASE_MANAGER_ID, make_payload) + assert second.status == Status.SUCCESS + + entries = _ledger_entries(dl) + assert len(entries) == 2, "each receipt commits its own entry" + assert ( + entries[0].payload_snapshot["object"]["rmState"] == RM.VALID.name + ), "run 1 records the adjudicated rm" + assert ( + entries[1].payload_snapshot["object"]["rmState"] + == RM.RECEIVED.name + ), ( + "run 2 adjudicated nothing, so a stale override from run 1 must not" + " rewrite its snapshot" + ) + + def test_a_distinct_status_id_does_not_inherit_the_override( + self, dl, make_payload + ): + """A leftover override for another object is ignored by the ID match.""" + current = _current_status(RM.VALID, CS_vfd.Vfd, CS_pxa.pxa) + asserted = _asserted_status(RM.RECEIVED, CS_vfd.VFd, CS_pxa.Pxa) + _seed_case(dl, current, asserted) + + assert ( + _run_tree(dl, asserted, CASE_MANAGER_ID, make_payload).status + == Status.SUCCESS + ) + + # Wholly acceptable, so the filter publishes nothing of its own. + second_status = _asserted_status( + RM.ACCEPTED, CS_vfd.VFd, CS_pxa.Pxa, status_id=SECOND_STATUS_ID + ) + dl.create(second_status) + assert ( + _run_tree(dl, second_status, CASE_MANAGER_ID, make_payload).status + == Status.SUCCESS + ) + + entries = _ledger_entries(dl) + assert len(entries) == 2 + second_snap = entries[1].payload_snapshot["object"] + assert second_snap["id"] == SECOND_STATUS_ID + assert ( + second_snap["rmState"] == RM.ACCEPTED.name + ), "the second status must be snapshotted as asserted" + + def test_filter_clears_a_stale_override_when_no_datalayer_is_available( + self, + ): + """The datalayer-missing early return must still clear both keys. + + ``update()`` clears before it checks for the DataLayer, so a node that + cannot do its job leaves no adjudication behind for the commit node to + act on. + """ + reader = py_trees.blackboard.Client(name="override-reader") + for key in (BB_LEDGER_PAYLOAD_OBJECT_OVERRIDE, BB_DIMENSION_FILTER): + reader.register_key(key=key, access=py_trees.common.Access.READ) + + node = FilterParticipantStatusDimensionsNode( + participant_id=PARTICIPANT_ID, status_id=ASSERTED_STATUS_ID + ) + node.setup() + node.blackboard.set( + BB_LEDGER_PAYLOAD_OBJECT_OVERRIDE, + {"object_id": ASSERTED_STATUS_ID, "fields": {"rmState": "CLOSED"}}, + overwrite=True, + ) + node.blackboard.set( + BB_DIMENSION_FILTER, + {"status_id": ASSERTED_STATUS_ID, "refused": ("rm",)}, + overwrite=True, + ) + + assert node.datalayer is None + assert node.update() == Status.FAILURE + assert reader.get(BB_LEDGER_PAYLOAD_OBJECT_OVERRIDE) is None + assert reader.get(BB_DIMENSION_FILTER) is None + + +class TestResolveDimensionFilter: + """The downstream helper must not read another execution's outcome.""" + + def test_returns_none_for_a_mismatched_status_id(self): + client = py_trees.blackboard.Client(name="filter-writer") + client.register_key( + key=BB_DIMENSION_FILTER, access=py_trees.common.Access.WRITE + ) + payload = {"status_id": ASSERTED_STATUS_ID, "refused": ("rm",)} + client.set(BB_DIMENSION_FILTER, payload, overwrite=True) + + assert resolve_dimension_filter(client, ASSERTED_STATUS_ID) is payload + assert resolve_dimension_filter(client, SECOND_STATUS_ID) is None + + def test_returns_none_when_unset_or_not_a_dict(self): + client = py_trees.blackboard.Client(name="filter-reader") + client.register_key( + key=BB_DIMENSION_FILTER, access=py_trees.common.Access.WRITE + ) + assert resolve_dimension_filter(client, ASSERTED_STATUS_ID) is None + + client.set(BB_DIMENSION_FILTER, None, overwrite=True) + assert resolve_dimension_filter(client, ASSERTED_STATUS_ID) is None + + +class TestMergeSnapshotObjectFields: + """Unit coverage for the patch merge applied to a payload snapshot.""" + + def test_patches_flat_fields_and_drops_stale_snake_case_twins(self): + merged = _merge_snapshot_object_fields( + { + "id": ASSERTED_STATUS_ID, + "rmState": "RECEIVED", + "rm_state": "RECEIVED", + "emConsentState": "SIGNATORY", + "name": "RECEIVED VFd", + }, + {"rmState": "VALID", "vfdState": "VFd"}, + ) + assert merged["rmState"] == "VALID" + assert merged["vfdState"] == "VFd" + assert "rm_state" not in merged, ( + "a stale snake_case twin would let a consumer read the value the" + " receiver just refused" + ) + assert merged["emConsentState"] == "SIGNATORY" + assert merged["id"] == ASSERTED_STATUS_ID + assert "name" not in merged, "the sender's derived label is dropped" + + def test_merges_one_level_of_nesting_without_replacing_it(self): + merged = _merge_snapshot_object_fields( + { + "id": ASSERTED_STATUS_ID, + "caseStatus": { + "id": f"{ASSERTED_STATUS_ID}/cs", + "type": "CaseStatus", + "pxaState": "PXA", + "pxa_state": "PXA", + "emState": "NONE", + }, + }, + {"caseStatus": {"pxaState": "pxa", "emState": "NONE"}}, + ) + case_status = merged["caseStatus"] + assert case_status["pxaState"] == "pxa" + assert "pxa_state" not in case_status + assert case_status["id"] == f"{ASSERTED_STATUS_ID}/cs" + assert case_status["type"] == "CaseStatus" + + def test_leaves_a_bare_reference_alone(self): + """Clobbering a reference string would drop the reference entirely.""" + current = { + "id": ASSERTED_STATUS_ID, + "caseStatus": f"{ASSERTED_STATUS_ID}/cs", + } + merged = _merge_snapshot_object_fields( + current, {"rmState": "VALID", "caseStatus": {"pxaState": "pxa"}} + ) + assert merged["caseStatus"] == f"{ASSERTED_STATUS_ID}/cs" + assert merged["rmState"] == "VALID" diff --git a/test/core/behaviors/sync/nodes/test_effects.py b/test/core/behaviors/sync/nodes/test_participant_status_effect.py similarity index 98% rename from test/core/behaviors/sync/nodes/test_effects.py rename to test/core/behaviors/sync/nodes/test_participant_status_effect.py index 84a9d74f9..37c1f53c9 100644 --- a/test/core/behaviors/sync/nodes/test_effects.py +++ b/test/core/behaviors/sync/nodes/test_participant_status_effect.py @@ -1,5 +1,5 @@ #!/usr/bin/env python -"""Regression tests for ApplyParticipantStatusFromLedgerNode (effects.py). +"""Regression tests for ApplyParticipantStatusFromLedgerNode. Covers the critical round-trip serialization bug: a CORE ParticipantStatus appended directly to as_CaseParticipant.participant_statuses was serialized with @@ -27,7 +27,7 @@ ) from vultron.adapters.driven.datalayer_sqlite import SqliteDataLayer from vultron.core.behaviors.bridge import BTBridge -from vultron.core.behaviors.sync.nodes.effects import ( +from vultron.core.behaviors.sync.nodes.participant_status_effect import ( ApplyParticipantStatusFromLedgerNode, ) from vultron.core.models.case_actor import VultronCaseActor diff --git a/test/core/states/test_cs_monotonic_predicates.py b/test/core/states/test_cs_monotonic_predicates.py new file mode 100644 index 000000000..b9ac3ee00 --- /dev/null +++ b/test/core/states/test_cs_monotonic_predicates.py @@ -0,0 +1,233 @@ +#!/usr/bin/env python + +# Copyright (c) 2026 Carnegie Mellon University and Contributors. +# - see Contributors.md for a full list of Contributors +# - see ContributionInstructions.md for information on how you can Contribute to this project +# Vultron Multiparty Coordinated Vulnerability Disclosure Protocol Prototype is +# licensed under a MIT (SEI)-style license, please see LICENSE.md distributed +# with this Software or contact permission@sei.cmu.edu for full terms. +# Created, in part, with funding and support from the United States Government +# (see Acknowledgments file). This program may include and/or can make use of +# certain third party source code, object code, documentation and other files +# ("Third Party Software"). See LICENSE.md for more details. +# Carnegie Mellon®, CERT® and CERT Coordination Center® are registered in the +# U.S. Patent and Trademark Office by Carnegie Mellon University + +"""Exhaustive tests for the vfd/pxa monotone-forward predicates. + +:func:`~vultron.core.states.cs.is_monotonic_vfd_forward` and +:func:`~vultron.core.states.cs.is_monotonic_pxa_forward` are the weaker +companions to the adjacency checks: a peer may report a state several steps +ahead of the one the receiver holds (a vendor that became aware, readied and +deployed a fix between two status updates reports ``vfd → VFD`` in one +message), and that is monotone but not an adjacent transition. + +They are the acceptance rule for the ``vfd`` and ``pxa`` dimensions of a +received ``ParticipantStatus`` (RSH-05), so every ordered pair of states is +covered here rather than the handful the adjudication tests happen to exercise. + +Both state groups are tuples of independent one-way latches, so the expected +answer is a *strict superset* test over the set-components: monotone forward iff +``dest`` sets at least one component that ``source`` had not set and un-sets +none. Expressing the oracle as a bitmask subset test keeps it independent of +the position-wise component comparison the implementation uses. +""" + +import itertools + +import pytest + +from vultron.core.states.cs import ( + CS_pxa, + CS_vfd, + is_monotonic_pxa_forward, + is_monotonic_vfd_forward, +) + +# --------------------------------------------------------------------------- +# Oracle +# --------------------------------------------------------------------------- + + +def _latch_bits(member: CS_vfd | CS_pxa) -> int: + """Encode a state's set components as a bitmask. + + Each component of a ``VfdState``/``PxaState`` is a two-valued ``StrEnum`` + whose "set" value is spelled with an uppercase letter (``V``, ``F``, ``D``; + ``P``, ``X``, ``A``). The member *name* carries exactly that information, + so the mask is read off the name rather than the ``NamedTuple``. + """ + return sum( + 1 << index + for index, letter in enumerate(member.name) + if letter.isupper() + ) + + +def _expected_monotonic_forward( + source: CS_vfd | CS_pxa, dest: CS_vfd | CS_pxa +) -> bool: + """``dest`` is a strict superset of ``source``'s set components.""" + source_bits, dest_bits = _latch_bits(source), _latch_bits(dest) + return source_bits != dest_bits and (source_bits & ~dest_bits) == 0 + + +# --------------------------------------------------------------------------- +# vfd +# --------------------------------------------------------------------------- + + +class TestIsMonotonicVfdForward: + ALL = list(CS_vfd) + + @pytest.mark.parametrize( + "source,dest", + list(itertools.product(list(CS_vfd), repeat=2)), + ids=lambda m: m.name, + ) + def test_every_ordered_pair(self, source, dest): + assert is_monotonic_vfd_forward( + source, dest + ) is _expected_monotonic_forward(source, dest) + + @pytest.mark.parametrize("state", ALL, ids=lambda m: m.name) + def test_equality_is_not_forward(self, state): + """A status confirmation advances nothing; callers test equality.""" + assert is_monotonic_vfd_forward(state, state) is False + + @pytest.mark.parametrize( + "source,dest", + [ + (CS_vfd.vfd, CS_vfd.Vfd), + (CS_vfd.Vfd, CS_vfd.VFd), + (CS_vfd.VFd, CS_vfd.VFD), + ], + ids=["vfd->Vfd", "Vfd->VFd", "VFd->VFD"], + ) + def test_adjacent_steps_are_forward(self, source, dest): + assert is_monotonic_vfd_forward(source, dest) is True + + @pytest.mark.parametrize( + "source,dest", + [ + (CS_vfd.vfd, CS_vfd.VFd), + (CS_vfd.vfd, CS_vfd.VFD), + (CS_vfd.Vfd, CS_vfd.VFD), + ], + ids=["vfd->VFd", "vfd->VFD", "Vfd->VFD"], + ) + def test_multi_step_jumps_are_forward(self, source, dest): + """The whole point: adjacency is too strict for a peer's snapshot.""" + assert is_monotonic_vfd_forward(source, dest) is True + + @pytest.mark.parametrize( + "source,dest", + [ + (CS_vfd.VFD, CS_vfd.VFd), + (CS_vfd.VFd, CS_vfd.Vfd), + (CS_vfd.Vfd, CS_vfd.vfd), + (CS_vfd.VFD, CS_vfd.vfd), + ], + ids=["VFD->VFd", "VFd->Vfd", "Vfd->vfd", "VFD->vfd"], + ) + def test_regressions_are_refused(self, source, dest): + assert is_monotonic_vfd_forward(source, dest) is False + + def test_forward_pair_count(self): + """4 states on a single chain → 6 strictly-forward ordered pairs.""" + forward = [ + (s, d) + for s, d in itertools.product(self.ALL, repeat=2) + if is_monotonic_vfd_forward(s, d) + ] + assert len(forward) == 6 + + def test_relation_is_antisymmetric(self): + for s, d in itertools.product(self.ALL, repeat=2): + if is_monotonic_vfd_forward(s, d): + assert not is_monotonic_vfd_forward(d, s) + + +# --------------------------------------------------------------------------- +# pxa +# --------------------------------------------------------------------------- + + +class TestIsMonotonicPxaForward: + ALL = list(CS_pxa) + + @pytest.mark.parametrize( + "source,dest", + list(itertools.product(list(CS_pxa), repeat=2)), + ids=lambda m: m.name, + ) + def test_every_ordered_pair(self, source, dest): + assert is_monotonic_pxa_forward( + source, dest + ) is _expected_monotonic_forward(source, dest) + + @pytest.mark.parametrize("state", ALL, ids=lambda m: m.name) + def test_equality_is_not_forward(self, state): + assert is_monotonic_pxa_forward(state, state) is False + + @pytest.mark.parametrize( + "source,dest", + [ + (CS_pxa.pxa, CS_pxa.Pxa), + (CS_pxa.pxa, CS_pxa.pXa), + (CS_pxa.pxa, CS_pxa.pxA), + (CS_pxa.pxa, CS_pxa.PXA), + (CS_pxa.Pxa, CS_pxa.PXa), + (CS_pxa.pXa, CS_pxa.PXA), + ], + ids=[ + "pxa->Pxa", + "pxa->pXa", + "pxa->pxA", + "pxa->PXA", + "Pxa->PXa", + "pXa->PXA", + ], + ) + def test_independent_latches_may_set_in_any_combination( + self, source, dest + ): + """P/X/A are mutually independent, so any newly-set subset is forward.""" + assert is_monotonic_pxa_forward(source, dest) is True + + @pytest.mark.parametrize( + "source,dest", + [ + (CS_pxa.Pxa, CS_pxa.pxa), + (CS_pxa.PxA, CS_pxa.PXa), + (CS_pxa.PXA, CS_pxa.pxa), + (CS_pxa.PXa, CS_pxa.pXA), + ], + ids=["Pxa->pxa", "PxA->PXa", "PXA->pxa", "PXa->pXA"], + ) + def test_any_component_regression_refuses_the_whole_move( + self, source, dest + ): + """``PXa → pXA`` sets A but un-sets P — one latch reopening is enough.""" + assert is_monotonic_pxa_forward(source, dest) is False + + def test_forward_pair_count(self): + """3 independent latches → 3**3 subset pairs, minus the 2**3 equal.""" + forward = [ + (s, d) + for s, d in itertools.product(self.ALL, repeat=2) + if is_monotonic_pxa_forward(s, d) + ] + assert len(forward) == 3**3 - 2**3 + + def test_relation_is_antisymmetric(self): + for s, d in itertools.product(self.ALL, repeat=2): + if is_monotonic_pxa_forward(s, d): + assert not is_monotonic_pxa_forward(d, s) + + def test_relation_is_transitive(self): + for a, b, c in itertools.product(self.ALL, repeat=3): + if is_monotonic_pxa_forward(a, b) and is_monotonic_pxa_forward( + b, c + ): + assert is_monotonic_pxa_forward(a, c) diff --git a/vultron/core/behaviors/case/nodes/lifecycle.py b/vultron/core/behaviors/case/nodes/lifecycle.py index 77f5219af..bb702ff6c 100644 --- a/vultron/core/behaviors/case/nodes/lifecycle.py +++ b/vultron/core/behaviors/case/nodes/lifecycle.py @@ -40,6 +40,20 @@ logger = logging.getLogger(__name__) +#: Blackboard key by which a preceding read-only guard in any receive tree may +#: patch the ``object`` entry of the canonical ledger ``payload_snapshot`` read +#: by :class:`CommitCaseLedgerEntryNode`. The value is a mapping +#: ``{"object_id": , "fields": }``, +#: or ``None`` when no adjudication applies. +#: +#: ``fields`` is a *patch*, not a replacement object: the guard names only the +#: fields it adjudicated, keyed by their wire aliases, and they are merged onto +#: whatever the snapshot already holds. That keeps the recorded shape identical +#: to an unadjudicated entry's (RSH-05-009). +#: +#: Producers: :class:`~vultron.core.behaviors.status.nodes.dimension_filter.FilterParticipantStatusDimensionsNode`. +BB_LEDGER_PAYLOAD_OBJECT_OVERRIDE = "ledger_payload_object_override" + def _extract_payload_snapshot( activity: Any, dl: CasePersistence | None = None @@ -56,6 +70,63 @@ def _extract_payload_snapshot( ) +#: snake_case spellings of the patchable flat status fields. A snapshot is +#: normally serialized ``by_alias`` (camelCase), but a stale snake_case twin +#: left alongside a patched alias would let a consumer that prefers the +#: snake_case spelling read the value the receiver just refused. +_SNAKE_TWINS: dict[str, str] = { + "rmState": "rm_state", + "vfdState": "vfd_state", + "emState": "em_state", + "pxaState": "pxa_state", + "emConsentState": "em_consent_state", + "caseStatus": "case_status", +} + + +def _merge_snapshot_object_fields( + current: dict[str, Any], fields: dict[str, Any] +) -> dict[str, Any]: + """Merge an adjudication patch onto a snapshot ``object``. + + One level of nesting is merged rather than replaced so that patching + ``caseStatus.pxaState`` keeps the snapshot's ``caseStatus`` id and its other + fields. A ``caseStatus`` that is still a bare reference string is left + alone — there is nothing to merge into, and clobbering it would drop the + reference. + + ``name`` is dropped: it is a derived state summary and the sender's label + describes the value that was just refused. + """ + merged = dict(current) + for key, value in fields.items(): + existing = merged.get(key) + if isinstance(value, dict): + if not isinstance(existing, dict): + # Bare reference (or absent) — nothing to patch into. + continue + nested = dict(existing) + for nested_key, nested_value in value.items(): + nested[nested_key] = nested_value + nested.pop(_SNAKE_TWINS.get(nested_key, ""), None) + merged[key] = nested + continue + merged[key] = value + merged.pop(_SNAKE_TWINS.get(key, ""), None) + merged.pop("name", None) + return merged + + +def _snapshot_object_id(payload_snapshot: dict[str, Any]) -> str | None: + """Return the ID of a payload snapshot's ``object``, inlined or not.""" + value = payload_snapshot.get("object") + if isinstance(value, str): + return value or None + if isinstance(value, dict): + return value.get("id") or value.get("id_") or None + return None + + class CommitCaseLedgerEntryNode(DataLayerAction): """ Commit a hash-chained CaseLedgerEntry and fan it out to all case participants. @@ -107,6 +178,10 @@ def setup(self, **kwargs: Any) -> None: self.blackboard.register_key( key="sync_port", access=py_trees.common.Access.READ ) + self.blackboard.register_key( + key=BB_LEDGER_PAYLOAD_OBJECT_OVERRIDE, + access=py_trees.common.Access.READ, + ) def initialise(self) -> None: super().initialise() @@ -127,6 +202,47 @@ def _resolve_activity(self) -> Any | None: except KeyError: return None + def _resolve_payload_object_override( + self, payload_snapshot: dict[str, Any] + ) -> dict[str, Any] | None: + """Return a substitute ``object`` entry for the payload snapshot. + + A preceding read-only guard may have adjudicated the inbound assertion + and published the portion the receiver actually accepts (RSH-05). The + canonical entry must record *that*, not the raw claim, otherwise the + refused value is hash-chained and replicated to every participant. + + The override is a **patch**, not a replacement object: the guard names + only the fields it adjudicated and they are merged onto the snapshot's + existing ``object``. That keeps the snapshot in the same wire shape the + un-adjudicated path produces — flat ``rmState``/``vfdState``, nested + ``caseStatus``, ``@context``, ``emConsentState``, ``cvdRole`` — which + every replica and the invariant harness rely on (RSH-05-009, + CLP-07-001, CM-18-006). A whole-object replacement built in core would + instead emit core dimension objects, since core must not import the wire + layer to convert (ADR-0009, ADR-0017). + + The override names the object ID it applies to and is honoured only + when the snapshot's ``object`` refers to the same ID: the py_trees + blackboard is process-global and not cleared between executions, so an + unmatched override is a leftover from an earlier run and is ignored. + """ + try: + override = self.blackboard.get(BB_LEDGER_PAYLOAD_OBJECT_OVERRIDE) + except KeyError: + return None + if not isinstance(override, dict): + return None + fields = override.get("fields") + if not isinstance(fields, dict) or not fields: + return None + current = payload_snapshot.get("object") + if not isinstance(current, dict): + return None + if _snapshot_object_id(payload_snapshot) != override.get("object_id"): + return None + return _merge_snapshot_object_fields(current, fields) + def _activity_metadata( self, activity: Any | None, case_id: str ) -> tuple[str, str, dict[str, Any]]: @@ -179,6 +295,23 @@ def update(self) -> Status: payload_snapshot = dict(payload_snapshot) payload_snapshot["context"] = case_id + # Record the portion of the assertion the receiver accepts, when a + # preceding guard adjudicated it (RSH-05). + if payload_snapshot: + replacement = self._resolve_payload_object_override( + payload_snapshot + ) + if replacement is not None: + payload_snapshot = dict(payload_snapshot) + payload_snapshot["object"] = replacement + self.logger.info( + "%s: snapshotting the accepted portion of object '%s'" + " for case '%s' (RSH-05)", + self.name, + _snapshot_object_id(payload_snapshot), + case_id, + ) + tree = create_commit_log_entry_tree( case_id=case_id, object_id=object_id, diff --git a/vultron/core/behaviors/status/add_participant_status_tree.py b/vultron/core/behaviors/status/add_participant_status_tree.py index aa6c05ed9..58b699492 100644 --- a/vultron/core/behaviors/status/add_participant_status_tree.py +++ b/vultron/core/behaviors/status/add_participant_status_tree.py @@ -30,7 +30,7 @@ AddParticipantStatusBT (Sequence) ├─ VerifySenderIsParticipantNode # Step 1: sender must be known participant - ├─ CheckParticipantRMNotClosedNode # Guard: reject CLOSED→CLOSED rewrites + ├─ FilterParticipantStatusDimensionsNode # Guard: adjudicate rm/vfd/pxa separately (RSH-05) ├─ GuardedCommitOrSkip (Selector, only if case_id) # Record receipt first (CLP-10-006) │ ├─ Sequence("SkipIfNotCaseManager") │ │ └─ Inverter(CheckIsCaseManagerNode) @@ -41,8 +41,16 @@ │ └─ CaseOwnerApprovesStatusUpdate # Call-out: non-owners need approval └─ EmitAddCaseStatusToSelfNode # Seam 1 emit → triggers Seam 2 (RSH-01-003) +``FilterParticipantStatusDimensionsNode`` adjudicates ``rm``, ``vfd`` and +``pxa`` independently before the commit, so an unacceptable value in one +dimension no longer discards the accepted dimensions or aborts the Sequence +before the Seam 1 emit (RSH-05, ISSUE-2235). It replaces the former +``CheckParticipantRMNotClosedNode`` guard, subsuming the terminal-``RM.CLOSED`` +check: a wholly refused assertion still returns FAILURE here, before any +canonical ledger entry is committed. + Per specs/multi-actor-demo.yaml DEMOMA-07-003, DEMOMA-07-005. -Per specs/received-status-handling.yaml RSH-01-001 to RSH-01-004. +Per specs/received-status-handling.yaml RSH-01-001 to RSH-01-004, RSH-05. Per ADR-0050: canonical RM closure is routed through Leave(VulnerabilityCase) receive path in receive_close_case_tree, not here. """ @@ -65,8 +73,8 @@ append_participant_status_tree, ) from vultron.core.behaviors.status.nodes import ( - CheckParticipantRMNotClosedNode, EmitAddCaseStatusToSelfNode, + FilterParticipantStatusDimensionsNode, VerifySenderIsParticipantNode, ) from vultron.core.models.events.status import ( @@ -167,9 +175,10 @@ def add_participant_status_tree( sender_actor_id=actor_id, case_id=tree_case_id, ), - CheckParticipantRMNotClosedNode( + FilterParticipantStatusDimensionsNode( participant_id=participant_id, status_id=status_id, + status_obj_fallback=status_obj, ), ], effect_nodes=[ diff --git a/vultron/core/behaviors/status/append_participant_status_tree.py b/vultron/core/behaviors/status/append_participant_status_tree.py index fe7f0841d..4b5a56659 100644 --- a/vultron/core/behaviors/status/append_participant_status_tree.py +++ b/vultron/core/behaviors/status/append_participant_status_tree.py @@ -85,7 +85,10 @@ def append_participant_status_tree( status_id=status_id, status_obj_fallback=status_obj_fallback, ), - ValidateRMTransitionNode(participant_id=participant_id), + ValidateRMTransitionNode( + participant_id=participant_id, + status_id=status_id, + ), AppendStatusAndSaveParticipantNode( status_id=status_id, participant_id=participant_id, diff --git a/vultron/core/behaviors/status/nodes/__init__.py b/vultron/core/behaviors/status/nodes/__init__.py index 74c7fa6b4..38cec9764 100644 --- a/vultron/core/behaviors/status/nodes/__init__.py +++ b/vultron/core/behaviors/status/nodes/__init__.py @@ -25,10 +25,14 @@ and close-not-yet-emitted idempotency guard nodes - ``broadcast``: (removed — case-manager lookup consolidated into ``_resolve_case_manager_id`` in ``vultron.core.use_cases._helpers``) -- ``append``: Load, validate RM transition, and append action nodes +- ``dimension_filter``: Per-dimension partial-accept guard for inbound + ParticipantStatus (FilterParticipantStatusDimensionsNode, RSH-05) +- ``append``: Load, resolve and append action nodes (SkipIfIdempotentNode, LoadParticipantNode, CheckStatusNotAlreadyAppendedNode, ResolveAndPersistStatusObjectNode, - ValidateRMTransitionNode, AppendStatusAndSaveParticipantNode) + AppendStatusAndSaveParticipantNode) +- ``rm_validation``: All-or-nothing RM guards for the append sequence + (ValidateRMTransitionNode, CheckParticipantRMNotClosedNode) - ``lifecycle``: Public disclosure and auto-close emit lifecycle nodes (_PublicDisclosureSkipConditionNode, PublicDisclosureBranchNode, ThreatTerminationBranchNode, EmitAddCaseStatusToSelfNode, EmitCloseCaseNode) @@ -47,13 +51,19 @@ CloseNotYetEmittedConditionNode, VerifySenderIsParticipantNode, ) +from vultron.core.behaviors.status.nodes.dimension_filter import ( + BB_DIMENSION_FILTER, + FilterParticipantStatusDimensionsNode, +) from vultron.core.behaviors.status.nodes.append import ( AppendStatusAndSaveParticipantNode, - CheckParticipantRMNotClosedNode, CheckStatusNotAlreadyAppendedNode, LoadParticipantNode, ResolveAndPersistStatusObjectNode, SkipIfIdempotentNode, +) +from vultron.core.behaviors.status.nodes.rm_validation import ( + CheckParticipantRMNotClosedNode, ValidateRMTransitionNode, ) from vultron.core.behaviors.status.nodes.lifecycle import ( @@ -69,14 +79,18 @@ "AllParticipantsRMClosedConditionNode", "CloseNotYetEmittedConditionNode", "VerifySenderIsParticipantNode", + # dimension_filter + "BB_DIMENSION_FILTER", + "FilterParticipantStatusDimensionsNode", # append "LoadParticipantNode", "CheckStatusNotAlreadyAppendedNode", - "CheckParticipantRMNotClosedNode", "ResolveAndPersistStatusObjectNode", - "ValidateRMTransitionNode", "AppendStatusAndSaveParticipantNode", "SkipIfIdempotentNode", + # rm_validation + "CheckParticipantRMNotClosedNode", + "ValidateRMTransitionNode", # lifecycle "_PublicDisclosureSkipConditionNode", "PublicDisclosureBranchNode", diff --git a/vultron/core/behaviors/status/nodes/append.py b/vultron/core/behaviors/status/nodes/append.py index 2f66a2313..3d7c71ed5 100644 --- a/vultron/core/behaviors/status/nodes/append.py +++ b/vultron/core/behaviors/status/nodes/append.py @@ -15,9 +15,10 @@ """Append-participant-status leaf nodes for DEMOMA-07-003 step 2. -Contains the five leaf nodes that implement the append sequence: -load participant, check idempotency, resolve status object, validate RM -transition, and append + save. +Contains the leaf nodes that implement the append sequence: check idempotency, +load participant, resolve status object, and append + save. The RM-transition +guards that also participate in that sequence live in +:mod:`vultron.core.behaviors.status.nodes.rm_validation` (BTND-07-004). """ import logging @@ -26,19 +27,14 @@ import py_trees from py_trees.common import Status -from vultron.core.behaviors.helpers import ( - DataLayerAction, - DataLayerCondition, - read_rm_states, +from vultron.core.behaviors.helpers import DataLayerAction, DataLayerCondition +from vultron.core.behaviors.status.nodes.dimension_filter import ( + BB_DIMENSION_FILTER, + resolve_dimension_filter, ) from vultron.core.models.case_participant import CaseParticipant from vultron.core.models.participant_status import ParticipantStatus from vultron.core.models.protocols import PersistableModel -from vultron.core.states.rm import ( - RM, - is_monotonic_rm_forward, - is_valid_rm_transition, -) from vultron.core.models._helpers import _as_id logger = logging.getLogger(__name__) @@ -194,8 +190,14 @@ def update(self) -> Status: class ResolveAndPersistStatusObjectNode(DataLayerAction): """Resolve the status object by ID, persisting fallback if needed. - Tries the DataLayer first; if not found, uses ``status_obj_fallback``, - saves it, then re-reads the canonical wire-format record. + When :class:`~vultron.core.behaviors.status.nodes.dimension_filter.FilterParticipantStatusDimensionsNode` + has partially accepted the inbound status, the *filtered* status (refused + dimensions carried forward) is persisted at ``status_id`` and used in place + of the raw assertion, so that the appended record, the ledger ``object`` + reference and the Seam 2 emit all describe the accepted portion (RSH-05). + + Otherwise tries the DataLayer first; if not found, uses + ``status_obj_fallback``, saves it, then re-reads the canonical record. Validates that the resolved object is a ParticipantStatus (has rm and vfd attributes). @@ -223,13 +225,30 @@ def setup(self, **kwargs: Any) -> None: key="append_status_status_obj", access=py_trees.common.Access.WRITE, ) + self.blackboard.register_key( + key=BB_DIMENSION_FILTER, + access=py_trees.common.Access.READ, + ) def update(self) -> Status: if (f := self._require_datalayer()) is not None: return f assert self.datalayer is not None - status_obj = self.datalayer.read(self.status_id) + filtered = resolve_dimension_filter(self.blackboard, self.status_id) + if filtered is not None: + status_obj = filtered["filtered_status"] + self.datalayer.save(status_obj) + self.logger.info( + "ResolveAndPersistStatusObjectNode: persisted partially" + " accepted status '%s' (refused: %s) in place of the raw" + " assertion (RSH-05)", + self.status_id, + ", ".join(filtered["refused"]), + ) + status_obj = self.datalayer.read(self.status_id) or status_obj + else: + status_obj = self.datalayer.read(self.status_id) if not hasattr(status_obj, "id_"): status_obj = self.status_obj_fallback if status_obj is not None: @@ -264,104 +283,6 @@ def update(self) -> Status: return Status.SUCCESS -class ValidateRMTransitionNode(DataLayerCondition): - """Validate RM state transition rules. - - Checks that the new RM state does not violate transition rules: - - Accepts non-adjacent forward RM jumps (sender is authoritative) - - Rejects backwards RM transitions - - Returns SUCCESS if the transition is valid or if participant has no current - status (nothing to validate against). - - Returns FAILURE if a backwards RM transition is detected. - """ - - def __init__(self, participant_id: str, name: str | None = None): - super().__init__(name=name or self.__class__.__name__) - self.participant_id = participant_id - - def setup(self, **kwargs: Any) -> None: - super().setup(**kwargs) - self.blackboard.register_key( - key="append_status_participant", - access=py_trees.common.Access.READ, - ) - self.blackboard.register_key( - key="append_status_status_obj", - access=py_trees.common.Access.READ, - ) - - def update(self) -> Status: - participant = self.blackboard.get("append_status_participant") - status_obj = self.blackboard.get("append_status_status_obj") - - if participant is None or status_obj is None: - self.feedback_message = "Participant or status not on blackboard" - self.logger.warning( - "ValidateRMTransitionNode: %s", self.feedback_message - ) - return Status.FAILURE - - current_status = getattr(participant, "participant_status", None) - if current_status is None: - self.logger.debug("ValidateRMTransitionNode: no current status") - return Status.SUCCESS - - states = read_rm_states(self, status_obj, current_status) - if states is None: - return Status.FAILURE - new_rm_state, current_rm = states - if current_rm == RM.CLOSED: - self.feedback_message = ( - "Participant is already in terminal RM.CLOSED state" - f" (received {new_rm_state}) for participant" - f" '{self.participant_id}'" - ) - self.logger.info( - "ValidateRMTransitionNode: %s — rejecting", - self.feedback_message, - ) - return Status.FAILURE - - if current_rm == new_rm_state: - self.logger.debug( - "ValidateRMTransitionNode: no RM state change (both %s)", - current_rm, - ) - return Status.SUCCESS - - if is_valid_rm_transition(current_rm, new_rm_state): - self.logger.debug( - "ValidateRMTransitionNode: valid adjacent transition" - " %s → %s", - current_rm, - new_rm_state, - ) - return Status.SUCCESS - - if is_monotonic_rm_forward(current_rm, new_rm_state): - self.logger.info( - "ValidateRMTransitionNode: non-adjacent forward RM" - " transition %s → %s for participant '%s';" - " accepting sender-authoritative state", - current_rm, - new_rm_state, - self.participant_id, - ) - return Status.SUCCESS - - self.feedback_message = ( - f"Backwards RM transition {current_rm} → {new_rm_state}" - f" for participant '{self.participant_id}'" - ) - self.logger.warning( - "ValidateRMTransitionNode: %s — rejecting", - self.feedback_message, - ) - return Status.FAILURE - - class AppendStatusAndSaveParticipantNode(DataLayerAction): """Append the status object to the participant and save. @@ -417,84 +338,3 @@ def update(self) -> Status: self.participant_id, ) return Status.SUCCESS - - -class CheckParticipantRMNotClosedNode(DataLayerCondition): - """Pre-flight guard: FAILURE when participant is in RM.CLOSED with no prior - status match. - - Used in ``add_participant_status_tree`` precondition guards to reject - CLOSED→CLOSED rewrites before the commit runs (CLP-10-006). - - When ``status_id`` is supplied and the participant is CLOSED, returns - SUCCESS if ``status_id`` is already in ``participant.participant_statuses`` - (idempotent delivery of a VALID→CLOSED update whose trigger side already - appended the status). Returns FAILURE only for genuine CLOSED→CLOSED - rewrite attempts (status not yet in participant's list). - - Returns SUCCESS when the participant has no current status, the current - RM state is not CLOSED, or the incoming status was already appended. - """ - - def __init__( - self, - participant_id: str, - status_id: str = "", - name: str | None = None, - ) -> None: - super().__init__(name=name or self.__class__.__name__) - self.participant_id = participant_id - self.status_id = status_id - - def update(self) -> Status: - if (f := self._require_datalayer()) is not None: - return f - assert self.datalayer is not None - - participant = self.datalayer.read(self.participant_id) - if not isinstance(participant, CaseParticipant): - self.logger.debug( - "%s: participant '%s' not found — allowing (no terminal check)", - self.name, - self.participant_id, - ) - return Status.SUCCESS - - current_status = getattr(participant, "participant_status", None) - if current_status is None: - return Status.SUCCESS - - states = read_rm_states(self, current_status) - if states is None: - return Status.FAILURE - (current_rm,) = states - if current_rm != RM.CLOSED: - return Status.SUCCESS - - # Participant is CLOSED. Allow if the incoming status was already - # appended by the trigger side (idempotent re-delivery of VALID→CLOSED). - if self.status_id: - existing_ids = [ - _as_id(s) - for s in getattr(participant, "participant_statuses", []) - ] - if self.status_id in existing_ids: - self.logger.debug( - "%s: participant '%s' is CLOSED but status '%s' already" - " in participant_statuses — allowing idempotent commit", - self.name, - self.participant_id, - self.status_id, - ) - return Status.SUCCESS - - self.feedback_message = ( - f"Participant '{self.participant_id}' is already in terminal" - " RM.CLOSED — rejecting status update (DEMOMA-07-003)" - ) - self.logger.info( - "%s: %s", - self.name, - self.feedback_message, - ) - return Status.FAILURE diff --git a/vultron/core/behaviors/status/nodes/dimension_filter.py b/vultron/core/behaviors/status/nodes/dimension_filter.py new file mode 100644 index 000000000..b8b49b068 --- /dev/null +++ b/vultron/core/behaviors/status/nodes/dimension_filter.py @@ -0,0 +1,479 @@ +#!/usr/bin/env python + +# Copyright (c) 2026 Carnegie Mellon University and Contributors. +# - see Contributors.md for a full list of Contributors +# - see ContributionInstructions.md for information on how you can Contribute to this project +# Vultron Multiparty Coordinated Vulnerability Disclosure Protocol Prototype is +# licensed under a MIT (SEI)-style license, please see LICENSE.md distributed +# with this Software or contact permission@sei.cmu.edu for full terms. +# Created, in part, with funding and support from the United States Government +# (see Acknowledgments file). This program may include and/or can make use of +# certain third party source code, object code, documentation and other files +# ("Third Party Software"). See LICENSE.md for more details. +# Carnegie Mellon®, CERT® and CERT Coordination Center® are registered in the +# U.S. Patent and Trademark Office by Carnegie Mellon University + +"""Per-dimension partial-accept filtering for received ParticipantStatus. + +An inbound ``Add(ParticipantStatus, CaseParticipant)`` carries a snapshot of +several *independent* state machines: ``rm`` (Report Management), ``vfd`` +(vendor fix path), ``pxa`` (public state), ``em`` (embargo) and ``consent`` +(participant embargo consent). Because they are independent, a value that is +unacceptable in one dimension says nothing about the others. + +Before RSH-05, one refused dimension discarded the entire snapshot: the +receiving Case Actor dropped the accepted dimensions along with the refused +one and aborted the enclosing ``AddParticipantStatusBT`` Sequence, which also +skipped the Seam 1 → Seam 2 emit and therefore embargo teardown +(ISSUE-2235, RSH-01-003, RSH-01-004). + +:class:`FilterParticipantStatusDimensionsNode` adjudicates each dimension on +its own and publishes a *filtered* status in which refused dimensions carry +forward the participant's current value. It is a read-only precondition guard +(CLP-10-006): it reads the DataLayer but writes only to the blackboard, so it +runs *before* ``GuardedCommit`` and the canonical ledger entry can record the +accepted portion rather than the raw assertion. + +Per specs/received-status-handling.yaml RSH-05. +""" + +import logging +from typing import Any + +import py_trees +from py_trees.common import Status + +from vultron.core.behaviors.case.nodes.lifecycle import ( + BB_LEDGER_PAYLOAD_OBJECT_OVERRIDE, +) +from vultron.core.behaviors.helpers import DataLayerCondition +from vultron.core.models._helpers import _as_id +from vultron.core.models.case_participant import CaseParticipant +from vultron.core.models.dimensions import ( + PxaDimension, + RmDimension, + VfdDimension, +) +from vultron.core.models.participant_status import ParticipantStatus +from vultron.core.models.protocols import PersistableModel +from vultron.core.states.cs import ( + is_monotonic_pxa_forward, + is_monotonic_vfd_forward, +) +from vultron.core.states.rm import ( + RM, + is_monotonic_rm_forward, + is_valid_rm_transition, +) + +logger = logging.getLogger(__name__) + +#: Blackboard key carrying the per-dimension filter outcome for the append +#: nodes downstream (``ResolveAndPersistStatusObjectNode``, +#: ``ValidateRMTransitionNode``). ``None`` when nothing was filtered. +BB_DIMENSION_FILTER = "append_status_dimension_filter" + + +def _accepted_wire_patch(filtered: ParticipantStatus) -> dict[str, Any]: + """Return the adjudicated dimension values keyed by their wire aliases. + + The canonical ledger's ``payload_snapshot['object']`` is the *sender's* + wire-shaped ``ParticipantStatus`` — flat ``rmState``/``vfdState``, nested + ``caseStatus``, plus ``@context``, ``emConsentState`` and ``cvdRole``. The + override is therefore published as a **patch** rather than a replacement + object: dumping this core model would emit nested ``rm``/``vfd`` dimension + objects and lose the fields the guard never adjudicated, and core must not + import the wire layer to convert (ADR-0009, ADR-0017). Patching leaves the + snapshot's shape exactly as the non-override path produces it and rewrites + only what was adjudicated (RSH-05-004, RSH-05-009). + + The alias names below are the same ones the core models already accept as + wire-compat input — see ``ParticipantStatus._migrate_flat_fields`` and + ``CaseStatus._migrate_flat_fields`` — so they are part of core's existing + surface, not new knowledge of the wire format. + """ + patch: dict[str, Any] = { + "rmState": filtered.rm.state.name, + "vfdState": filtered.vfd.state.name, + } + if filtered.case_status is not None: + patch["caseStatus"] = { + "emState": filtered.case_status.em.state.name, + "pxaState": filtered.case_status.pxa.state.name, + } + return patch + + +def _to_core_status(status_obj: Any) -> ParticipantStatus | None: + """Return *status_obj* as a core :class:`ParticipantStatus`, or ``None``. + + ``SqliteDataLayer.read`` already returns core models, but the fallback + object supplied by the tree factory comes from the wire layer with flat + ``rmState``/``vfdState`` fields. The core model's ``_migrate_flat_fields`` + validator accepts that shape, so a dump-and-revalidate normalises both. + """ + if isinstance(status_obj, ParticipantStatus): + return status_obj + if status_obj is None or not hasattr(status_obj, "model_dump"): + return None + try: + return ParticipantStatus.model_validate( + status_obj.model_dump( + mode="json", + by_alias=True, + serialize_as_any=True, + exclude_none=True, + ) + ) + except Exception as exc: # pragma: no cover - defensive + logger.warning( + "FilterParticipantStatusDimensionsNode: could not normalise" + " status object '%s' to a core ParticipantStatus: %s", + _as_id(status_obj), + exc, + ) + return None + + +def _significant_state(status: ParticipantStatus) -> tuple: + """Return the protocol-significant fields of *status* as a comparable tuple. + + Used to decide whether a filtered status still carries information the + case does not already hold. Identity fields (``id``, timestamps, ``name``) + are deliberately excluded — a status that merely restates the participant's + current state under a new ID is not new information. + """ + case_status = status.case_status + return ( + status.rm.state, + status.vfd.state, + None if case_status is None else case_status.em.state, + None if case_status is None else case_status.pxa.state, + None if status.consent is None else status.consent.state, + status.case_engagement, + status.embargo_adherence, + tuple(sorted(str(role) for role in status.cvd_role)), + ) + + +def _dimension_state(status: ParticipantStatus, dimension: str) -> Any: + """Return the state of one adjudicated dimension of *status*. + + Used to tell a dimension that was genuinely *rewritten* from one that was + blocked but whose recorded value matches the assertion anyway. + """ + if dimension == "rm": + return status.rm.state + if dimension == "vfd": + return status.vfd.state + if dimension == "pxa": + return ( + None + if status.case_status is None + else status.case_status.pxa.state + ) + return None + + +def _rm_is_acceptable(current: RM, asserted: RM) -> bool: + """Return True if *asserted* is an acceptable RM value given *current*. + + ``RM.CLOSED`` is terminal (DEMOMA-07-003): once a participant has closed, + no further RM value — not even ``CLOSED`` again — is acceptable. Otherwise + a status confirmation (no change), a valid adjacent transition, or a + non-adjacent but monotone forward jump are all acceptable; the sender is + authoritative about its own RM progress. + """ + if current == RM.CLOSED: + return False + if asserted == current: + return True + return is_valid_rm_transition( + current, asserted + ) or is_monotonic_rm_forward(current, asserted) + + +def _adjudicate_dimensions( + current: ParticipantStatus, asserted: ParticipantStatus +) -> tuple[list[str], dict[str, Any]]: + """Adjudicate ``rm``, ``vfd`` and ``pxa`` independently. + + Returns the names of the refused dimensions and the ``model_copy`` update + that carries the current value forward for each of them. ``em``, + ``consent``, ``case_engagement``, ``embargo_adherence``, ``cvd_role`` and + ``tracking_id`` are not adjudicated here — ``em`` in particular belongs to + Seam 2 (ADR-0046, ISSUE-2256). + + The two return values are deliberately not the same set. ``refused`` names + the dimensions whose *asserted* value was rejected; ``update_fields`` also + carries dimensions the sender said nothing about, which must be preserved + rather than dropped. An inbound status with no ``case_status`` at all is + the common case: it asserts nothing about ``pxa``/``em``, so the + participant's current ``case_status`` is carried forward instead of letting + the omission erase state the receiver already holds (RSH-05-002). + """ + refused: list[str] = [] + update_fields: dict[str, Any] = {} + + if not _rm_is_acceptable(current.rm.state, asserted.rm.state): + refused.append("rm") + update_fields["rm"] = RmDimension(state=current.rm.state) + + current_vfd = current.vfd.state + asserted_vfd = asserted.vfd.state + if asserted_vfd != current_vfd and not is_monotonic_vfd_forward( + current_vfd, asserted_vfd + ): + refused.append("vfd") + update_fields["vfd"] = VfdDimension(state=current_vfd) + + asserted_cs = asserted.case_status + current_cs = current.case_status + if asserted_cs is None and current_cs is not None: + # Nothing asserted about pxa/em — carry the receiver's own view + # forward. Persisting the assertion as-is would blank both. + update_fields["case_status"] = current_cs.model_copy(deep=True) + elif asserted_cs is not None and current_cs is not None: + current_pxa = current_cs.pxa.state + asserted_pxa = asserted_cs.pxa.state + if asserted_pxa != current_pxa and not is_monotonic_pxa_forward( + current_pxa, asserted_pxa + ): + refused.append("pxa") + update_fields["case_status"] = asserted_cs.model_copy( + update={"pxa": PxaDimension(state=current_pxa)} + ) + + return refused, update_fields + + +class FilterParticipantStatusDimensionsNode(DataLayerCondition): + """Adjudicate each dimension of an inbound ParticipantStatus separately. + + Read-only precondition guard (CLP-10-006): reads the participant and the + asserted status from the DataLayer and writes only to the blackboard. + + For each of ``rm``, ``vfd`` and ``pxa`` the asserted value is accepted when + it confirms or monotonically advances the participant's current value, and + refused otherwise. Refused dimensions carry forward the current value into + a *filtered* status which is published on the blackboard for the append + nodes and, as a serialized ``object`` override, for the canonical ledger + commit. ``em``, ``consent``, ``case_engagement``, ``embargo_adherence``, + ``cvd_role`` and ``tracking_id`` pass through untouched — ``em`` in + particular is Seam 2's to adjudicate (ADR-0046, ISSUE-2256). + + Returns: + SUCCESS when there is nothing to filter (no participant, no current + status, idempotent re-delivery, or every dimension acceptable) and when + a partial accept was computed. + + FAILURE only when at least one dimension was refused *and* the + resulting filtered status is indistinguishable from the participant's + current state — the assertion carried no acceptable information, so + there is nothing to record and no ledger entry should be committed. + + Per specs/received-status-handling.yaml RSH-05. + """ + + def __init__( + self, + participant_id: str, + status_id: str, + status_obj_fallback: PersistableModel | None = None, + name: str | None = None, + ) -> None: + super().__init__(name=name or self.__class__.__name__) + self.participant_id = participant_id + self.status_id = status_id + self.status_obj_fallback = status_obj_fallback + + def setup(self, **kwargs: Any) -> None: + super().setup(**kwargs) + self.blackboard.register_key( + key=BB_DIMENSION_FILTER, + access=py_trees.common.Access.WRITE, + ) + self.blackboard.register_key( + key=BB_LEDGER_PAYLOAD_OBJECT_OVERRIDE, + access=py_trees.common.Access.WRITE, + ) + + def _publish( + self, + refused: tuple[str, ...], + filtered: ParticipantStatus | None, + ) -> None: + """Publish (or clear) the filter outcome on the blackboard. + + The py_trees blackboard is process-global and is not cleared between + executions, so both keys are written on *every* tick — including with + ``None`` when no filtering applies — to prevent a previous run's + override from leaking into this one. + """ + if filtered is None: + self.blackboard.set(BB_DIMENSION_FILTER, None, overwrite=True) + self.blackboard.set( + BB_LEDGER_PAYLOAD_OBJECT_OVERRIDE, None, overwrite=True + ) + return + + self.blackboard.set( + BB_DIMENSION_FILTER, + { + "status_id": self.status_id, + "participant_id": self.participant_id, + "refused": refused, + "filtered_status": filtered, + }, + overwrite=True, + ) + self.blackboard.set( + BB_LEDGER_PAYLOAD_OBJECT_OVERRIDE, + { + "object_id": self.status_id, + "fields": _accepted_wire_patch(filtered), + }, + overwrite=True, + ) + + def _resolve_asserted(self) -> ParticipantStatus | None: + """Return the asserted status as a core model, DataLayer first.""" + assert self.datalayer is not None + from_dl = ( + self.datalayer.read(self.status_id) if self.status_id else None + ) + return _to_core_status( + from_dl if from_dl is not None else self.status_obj_fallback + ) + + def update(self) -> Status: + # Clear first, unconditionally: the no-op paths below must not inherit + # a previous execution's override from the process-global blackboard, + # and neither must the datalayer-missing early return (BT-17-003/004). + self._publish((), None) + + if (f := self._require_datalayer()) is not None: + return f + assert self.datalayer is not None + + participant = self.datalayer.read(self.participant_id) + if not isinstance(participant, CaseParticipant): + # LoadParticipantNode reports the missing participant; nothing to + # filter against here. + self._publish((), None) + return Status.SUCCESS + + existing_ids = [ + _as_id(s) for s in getattr(participant, "participant_statuses", []) + ] + if self.status_id and self.status_id in existing_ids: + # Idempotent re-delivery: the status is already recorded, so the + # append subtree short-circuits and there is nothing to filter. + self._publish((), None) + return Status.SUCCESS + + current = getattr(participant, "participant_status", None) + asserted = self._resolve_asserted() + if not isinstance(current, ParticipantStatus) or asserted is None: + self._publish((), None) + return Status.SUCCESS + + refused, update_fields = _adjudicate_dimensions(current, asserted) + if not update_fields: + return Status.SUCCESS + + # ``name`` on a ParticipantStatus is a derived state summary (the wire + # model rebuilds it from the dimension names whenever it is ``None``). + # Carrying the sender's label forward would leave the recorded object + # describing itself by the refused value, so clear it and let it be + # regenerated from what was actually accepted. + update_fields["name"] = None + filtered = asserted.model_copy(update=update_fields) + + # RSH-05-005: nothing acceptable was carried by this assertion, so + # appending it would grow the status history and the hash chain without + # recording a state change. Reached both when every refused dimension + # left the snapshot at the current state and when an omitted + # ``case_status`` was the only thing carried forward. + if _significant_state(filtered) == _significant_state(current): + self.feedback_message = ( + f"Status '{self.status_id}' refused in full for participant" + f" '{self.participant_id}': {self._carry_summary(refused)}" + " and no other dimension carries new state" + ) + self.logger.info("%s: %s", self.name, self.feedback_message) + self._publish((), None) + return Status.FAILURE + + self._publish(tuple(refused), filtered) + self.feedback_message = ( + f"Partially accepted status '{self.status_id}' for participant" + f" '{self.participant_id}': {self._carry_summary(refused)}" + ) + # A refused dimension whose recorded value equals the asserted one + # discarded nothing — RM.CLOSED restated by a participant that has + # already closed is the common case (RSH-05-006). Naming it as a + # refusal in the operator-facing log would misdescribe the audit trail, + # so report what was actually rewritten. + rewritten = [ + dim + for dim in refused + if _dimension_state(filtered, dim) + != _dimension_state(asserted, dim) + ] + self.logger.warning( + "%s: %s for participant '%s' (asserted rm=%s vfd=%s pxa=%s;" + " recording rm=%s vfd=%s pxa=%s) — RSH-05 partial accept", + self.name, + ( + f"rewrote dimension(s) {', '.join(rewritten)}" + if rewritten + else "blocked dimension(s) " + + ", ".join(refused) + + " with no change to the asserted value" + ), + self.participant_id, + asserted.rm.state, + asserted.vfd.state, + ( + None + if asserted.case_status is None + else asserted.case_status.pxa.state + ), + filtered.rm.state, + filtered.vfd.state, + ( + None + if filtered.case_status is None + else filtered.case_status.pxa.state + ), + ) + return Status.SUCCESS + + @staticmethod + def _carry_summary(refused: list[str]) -> str: + """Describe what the filter did, for feedback messages.""" + if refused: + return f"refused dimension(s) {', '.join(refused)}" + return "carried the current case_status forward (none asserted)" + + +def resolve_dimension_filter( + blackboard: py_trees.blackboard.Client, status_id: str +) -> dict[str, Any] | None: + """Return the filter outcome for *status_id*, or ``None``. + + Helper for the append nodes downstream of + :class:`FilterParticipantStatusDimensionsNode`. The ``status_id`` match + guards against a stale entry from an earlier execution, since the py_trees + blackboard is process-global. + """ + try: + payload = blackboard.get(BB_DIMENSION_FILTER) + except KeyError: + return None + if not isinstance(payload, dict): + return None + if payload.get("status_id") != status_id: + return None + return payload diff --git a/vultron/core/behaviors/status/nodes/rm_validation.py b/vultron/core/behaviors/status/nodes/rm_validation.py new file mode 100644 index 000000000..4d382b852 --- /dev/null +++ b/vultron/core/behaviors/status/nodes/rm_validation.py @@ -0,0 +1,276 @@ +#!/usr/bin/env python + +# Copyright (c) 2026 Carnegie Mellon University and Contributors. +# - see Contributors.md for a full list of Contributors +# - see ContributionInstructions.md for information on how you can Contribute to this project +# Vultron Multiparty Coordinated Vulnerability Disclosure Protocol Prototype is +# licensed under a MIT (SEI)-style license, please see LICENSE.md distributed +# with this Software or contact permission@sei.cmu.edu for full terms. +# Created, in part, with funding and support from the United States Government +# (see Acknowledgments file). This program may include and/or can make use of +# certain third party source code, object code, documentation and other files +# ("Third Party Software"). See LICENSE.md for more details. +# Carnegie Mellon®, CERT® and CERT Coordination Center® are registered in the +# U.S. Patent and Trademark Office by Carnegie Mellon University + +"""RM-transition guards for the append-participant-status path. + +Both nodes here adjudicate the ``rm`` dimension *alone* and refuse the whole +snapshot when it is unacceptable. That is correct for the standalone +``append_participant_status_tree``, where the caller has already decided which +status to append, but it is the wrong shape for a receive-side seam — see +:mod:`vultron.core.behaviors.status.nodes.dimension_filter` and ADR-0061 +(RSH-05, ISSUE-2235). +""" + +import logging +from typing import Any + +import py_trees +from py_trees.common import Status + +from vultron.core.behaviors.helpers import DataLayerCondition, read_rm_states +from vultron.core.behaviors.status.nodes.dimension_filter import ( + BB_DIMENSION_FILTER, + resolve_dimension_filter, +) +from vultron.core.models._helpers import _as_id +from vultron.core.models.case_participant import CaseParticipant +from vultron.core.states.rm import ( + RM, + is_monotonic_rm_forward, + is_valid_rm_transition, +) + +logger = logging.getLogger(__name__) + + +class ValidateRMTransitionNode(DataLayerCondition): + """Validate RM state transition rules. + + Checks that the new RM state does not violate transition rules: + - Accepts non-adjacent forward RM jumps (sender is authoritative) + - Rejects backwards RM transitions + + When + :class:`~vultron.core.behaviors.status.nodes.dimension_filter.FilterParticipantStatusDimensionsNode` + has already refused the ``rm`` dimension and carried the participant's + current value forward, this node accepts that value: the transition was + adjudicated upstream and re-rejecting it here would abort the Sequence and + discard the dimensions that *were* accepted (RSH-05, ISSUE-2235). The + checks below still apply when the node is used standalone, without the + filter — which is how ``append_participant_status_tree`` is exercised + directly. + + Returns SUCCESS if the transition is valid or if participant has no current + status (nothing to validate against). + + Returns FAILURE if a backwards RM transition is detected. + """ + + def __init__( + self, + participant_id: str, + status_id: str = "", + name: str | None = None, + ): + super().__init__(name=name or self.__class__.__name__) + self.participant_id = participant_id + self.status_id = status_id + + def setup(self, **kwargs: Any) -> None: + super().setup(**kwargs) + self.blackboard.register_key( + key="append_status_participant", + access=py_trees.common.Access.READ, + ) + self.blackboard.register_key( + key="append_status_status_obj", + access=py_trees.common.Access.READ, + ) + self.blackboard.register_key( + key=BB_DIMENSION_FILTER, + access=py_trees.common.Access.READ, + ) + + def _rm_was_carried_forward( + self, current_rm: RM, new_rm_state: RM + ) -> bool: + """Return True if the filter node refused ``rm`` and carried *current*. + + Only a filtered status that actually restates the current RM value is + honoured; anything else falls through to the normal checks. + """ + if not self.status_id: + return False + filtered = resolve_dimension_filter(self.blackboard, self.status_id) + if filtered is None or "rm" not in filtered["refused"]: + return False + return current_rm == new_rm_state + + def update(self) -> Status: + participant = self.blackboard.get("append_status_participant") + status_obj = self.blackboard.get("append_status_status_obj") + + if participant is None or status_obj is None: + self.feedback_message = "Participant or status not on blackboard" + self.logger.warning( + "ValidateRMTransitionNode: %s", self.feedback_message + ) + return Status.FAILURE + + current_status = getattr(participant, "participant_status", None) + if current_status is None: + self.logger.debug("ValidateRMTransitionNode: no current status") + return Status.SUCCESS + + states = read_rm_states(self, status_obj, current_status) + if states is None: + return Status.FAILURE + new_rm_state, current_rm = states + if self._rm_was_carried_forward(current_rm, new_rm_state): + self.logger.debug( + "ValidateRMTransitionNode: rm was refused upstream and" + " carried forward as %s — accepting (RSH-05)", + current_rm, + ) + return Status.SUCCESS + + if current_rm == RM.CLOSED: + self.feedback_message = ( + "Participant is already in terminal RM.CLOSED state" + f" (received {new_rm_state}) for participant" + f" '{self.participant_id}'" + ) + self.logger.info( + "ValidateRMTransitionNode: %s — rejecting", + self.feedback_message, + ) + return Status.FAILURE + + if current_rm == new_rm_state: + self.logger.debug( + "ValidateRMTransitionNode: no RM state change (both %s)", + current_rm, + ) + return Status.SUCCESS + + if is_valid_rm_transition(current_rm, new_rm_state): + self.logger.debug( + "ValidateRMTransitionNode: valid adjacent transition" + " %s → %s", + current_rm, + new_rm_state, + ) + return Status.SUCCESS + + if is_monotonic_rm_forward(current_rm, new_rm_state): + self.logger.info( + "ValidateRMTransitionNode: non-adjacent forward RM" + " transition %s → %s for participant '%s';" + " accepting sender-authoritative state", + current_rm, + new_rm_state, + self.participant_id, + ) + return Status.SUCCESS + + self.feedback_message = ( + f"Backwards RM transition {current_rm} → {new_rm_state}" + f" for participant '{self.participant_id}'" + ) + self.logger.warning( + "ValidateRMTransitionNode: %s — rejecting", + self.feedback_message, + ) + return Status.FAILURE + + +class CheckParticipantRMNotClosedNode(DataLayerCondition): + """Pre-flight guard: FAILURE when participant is in RM.CLOSED with no prior + status match. + + .. deprecated:: RSH-05 + + Superseded by + :class:`~vultron.core.behaviors.status.nodes.dimension_filter.FilterParticipantStatusDimensionsNode`, + which subsumes this terminal-``RM.CLOSED`` check and no longer discards + the other dimensions of the snapshot along with ``rm``. Do not wire + this node back into ``add_participant_status_tree`` — an all-or-nothing + RM guard there is exactly the defect in ISSUE-2235. Retained for + callers that want the narrow check in isolation. + + Rejects CLOSED→CLOSED rewrites before the commit runs (CLP-10-006). + + When ``status_id`` is supplied and the participant is CLOSED, returns + SUCCESS if ``status_id`` is already in ``participant.participant_statuses`` + (idempotent delivery of a VALID→CLOSED update whose trigger side already + appended the status). Returns FAILURE only for genuine CLOSED→CLOSED + rewrite attempts (status not yet in participant's list). + + Returns SUCCESS when the participant has no current status, the current + RM state is not CLOSED, or the incoming status was already appended. + """ + + def __init__( + self, + participant_id: str, + status_id: str = "", + name: str | None = None, + ) -> None: + super().__init__(name=name or self.__class__.__name__) + self.participant_id = participant_id + self.status_id = status_id + + def update(self) -> Status: + if (f := self._require_datalayer()) is not None: + return f + assert self.datalayer is not None + + participant = self.datalayer.read(self.participant_id) + if not isinstance(participant, CaseParticipant): + self.logger.debug( + "%s: participant '%s' not found — allowing (no terminal check)", + self.name, + self.participant_id, + ) + return Status.SUCCESS + + current_status = getattr(participant, "participant_status", None) + if current_status is None: + return Status.SUCCESS + + states = read_rm_states(self, current_status) + if states is None: + return Status.FAILURE + (current_rm,) = states + if current_rm != RM.CLOSED: + return Status.SUCCESS + + # Participant is CLOSED. Allow if the incoming status was already + # appended by the trigger side (idempotent re-delivery of VALID→CLOSED). + if self.status_id: + existing_ids = [ + _as_id(s) + for s in getattr(participant, "participant_statuses", []) + ] + if self.status_id in existing_ids: + self.logger.debug( + "%s: participant '%s' is CLOSED but status '%s' already" + " in participant_statuses — allowing idempotent commit", + self.name, + self.participant_id, + self.status_id, + ) + return Status.SUCCESS + + self.feedback_message = ( + f"Participant '{self.participant_id}' is already in terminal" + " RM.CLOSED — rejecting status update (DEMOMA-07-003)" + ) + self.logger.info( + "%s: %s", + self.name, + self.feedback_message, + ) + return Status.FAILURE diff --git a/vultron/core/behaviors/sync/nodes/__init__.py b/vultron/core/behaviors/sync/nodes/__init__.py index 1374fa1fd..fc5d17b9d 100644 --- a/vultron/core/behaviors/sync/nodes/__init__.py +++ b/vultron/core/behaviors/sync/nodes/__init__.py @@ -26,6 +26,11 @@ - ``chain``: Chain reconstruction and log entry creation action nodes - ``canonical_entry``: Canonical ``payloadSnapshot`` validation (CLP-07) - ``replay``: Replay and fan-out action nodes for replication +- ``effects``: Ledger-apply side-effect nodes (note, invite-accept, close-case) +- ``participant_status_effect``: Ledger-apply of ``ParticipantStatus``, with the + monotonic-RM ratchet (ADR-0061) +- ``offer_report_effect``, ``ownership_effects``, ``ownership_offer_effect``: + per-effect ledger-apply nodes """ from vultron.core.behaviors.sync.nodes.chain import ( @@ -64,6 +69,8 @@ ApplyCloseCaseFromLedgerNode, ApplyInviteAcceptFromLedgerNode, ApplyNoteFromLedgerNode, +) +from vultron.core.behaviors.sync.nodes.participant_status_effect import ( ApplyParticipantStatusFromLedgerNode, ) from vultron.core.behaviors.sync.nodes.offer_report_effect import ( @@ -107,10 +114,12 @@ "IsSubmitReportEventNode", "IsOwnershipTransferEventNode", # effects - "ApplyParticipantStatusFromLedgerNode", "ApplyNoteFromLedgerNode", "ApplyInviteAcceptFromLedgerNode", "ApplyCloseCaseFromLedgerNode", + # participant_status_effect + "ApplyParticipantStatusFromLedgerNode", + # per-effect ledger-apply modules "ApplyOfferReportFromLedgerNode", "ApplyOwnershipTransferFromLedgerNode", "ApplyOfferOwnershipTransferFromLedgerNode", diff --git a/vultron/core/behaviors/sync/nodes/effects.py b/vultron/core/behaviors/sync/nodes/effects.py index a2f343987..37eeb7ba1 100644 --- a/vultron/core/behaviors/sync/nodes/effects.py +++ b/vultron/core/behaviors/sync/nodes/effects.py @@ -20,8 +20,6 @@ Currently implemented effects: -- :class:`ApplyParticipantStatusFromLedgerNode`: applies an - ``add_participant_status_to_participant`` event to the local participant record. - :class:`ApplyNoteFromLedgerNode`: applies an ``add_note_to_case`` event to the local case replica by attaching the note ID to ``notes``. - :class:`ApplyInviteAcceptFromLedgerNode`: applies an @@ -32,6 +30,12 @@ :class:`~vultron.core.models.participant_status.ParticipantStatus` to ``RM.CLOSED`` (CM-23-003, ADR-0050). +Effects that carry enough logic to warrant their own module live beside this +one and import :func:`_extract_id_from_field` from here: +:mod:`~vultron.core.behaviors.sync.nodes.participant_status_effect`, +:mod:`~vultron.core.behaviors.sync.nodes.ownership_effects` and +:mod:`~vultron.core.behaviors.sync.nodes.offer_report_effect` (BTND-07-004). + Per specs/multi-actor-demo.yaml DEMOMA-07-003 step 3, specs/case-management.yaml CM-23-003, and specs/sync-ledger-replication.yaml SYNC-02-002. @@ -40,7 +44,7 @@ from __future__ import annotations import logging -from typing import Any, cast +from typing import Any import py_trees from py_trees.common import Status @@ -50,14 +54,11 @@ from vultron.core.models.case import VulnerabilityCase from vultron.core.models.case_participant import CaseParticipant from vultron.core.models.participant_status import ( - ParticipantStatus, participant_status_rm_state, ) logger = logging.getLogger(__name__) -_ADD_PARTICIPANT_STATUS_EVENT = "add_participant_status_to_participant" - def _extract_id_from_field(value: Any) -> str | None: """Return the string ID from an AS2 object field. @@ -74,143 +75,6 @@ def _extract_id_from_field(value: Any) -> str | None: return getattr(value, "id_", None) or getattr(value, "id", None) or None -class ApplyParticipantStatusFromLedgerNode(DataLayerAction): - """Apply an ``add_participant_status_to_participant`` ledger entry locally. - - When a non-Case-Actor participant receives - ``Announce(CaseLedgerEntry)`` and the entry's ``event_type`` is - ``add_participant_status_to_participant``, this node reconstructs the - :class:`~vultron.core.models.participant_status.ParticipantStatus` from - the entry's ``payload_snapshot`` and appends it to the matching - :class:`~vultron.core.models.case_participant.CaseParticipant` in the - local DataLayer. - - The Case Actor is considered authoritative: RM-state validation is skipped - (the Case Actor already validated the transition before committing the - entry). Idempotency is preserved — if the status ID is already present in - the participant's list, the node returns SUCCESS without modifying the - DataLayer. - - Lenient on missing data: if the participant is not found in the local - DataLayer (this actor may have a partial view of the case), or the - payload snapshot is incomplete, the node returns SUCCESS without error to - avoid blocking the ``Announce`` processing flow. - - Per specs/multi-actor-demo.yaml DEMOMA-07-003 step 3, - specs/sync-ledger-replication.yaml SYNC-02-002. - """ - - def setup(self, **kwargs: Any) -> None: - super().setup(**kwargs) - self.blackboard.register_key( - key="activity", access=py_trees.common.Access.READ - ) - - def update(self) -> Status: - if (f := self._require_datalayer()) is not None: - return f - assert self.datalayer is not None - from vultron.core.behaviors.sync.nodes.conditions import ( - _require_log_entry, - ) - - entry = _require_log_entry(self.blackboard.activity, self.name) - snapshot = entry.payload_snapshot - - status_data = snapshot.get("object") - target_data = snapshot.get("target") - - status_id = _extract_id_from_field(status_data) - participant_id = _extract_id_from_field(target_data) - - if not status_id or not participant_id: - self.logger.debug( - "%s: payload_snapshot missing 'object' or 'target' id" - " — skipping status apply (non-fatal)", - self.name, - ) - return Status.SUCCESS - - participant = self.datalayer.read(participant_id) - if not isinstance(participant, CaseParticipant): - self.logger.debug( - "%s: participant '%s' not found in local DataLayer" - " — skipping (non-fatal, partial case view)", - self.name, - participant_id, - ) - return Status.SUCCESS - - existing_ids = [_as_id(s) for s in participant.participant_statuses] - if status_id in existing_ids: - self.logger.debug( - "%s: status '%s' already present on participant '%s'" - " — idempotent no-op", - self.name, - status_id, - participant_id, - ) - return Status.SUCCESS - - if not isinstance(status_data, dict): - self.logger.warning( - "%s: payload_snapshot 'object' is not a dict" - " — cannot reconstruct ParticipantStatus for '%s'", - self.name, - status_id, - ) - return Status.SUCCESS - - try: - status_obj = ParticipantStatus.model_validate(status_data) - except Exception as exc: - self.logger.warning( - "%s: failed to reconstruct ParticipantStatus from" - " payload_snapshot for '%s': %s", - self.name, - status_id, - exc, - ) - return Status.SUCCESS - - if self.datalayer.read(status_id) is None: - self.datalayer.save(status_obj) - - # Read back from the DataLayer to obtain the vocabulary-typed - # (wire-format) version of the status object. Appending the - # core-model instance directly to ``participant_statuses`` - # (typed ``list[WireParticipantStatus]``) causes Pydantic to - # serialize the list with default field values instead of the - # actual values, because the declared element type governs - # serialization when the runtime type differs. Reading back - # via the DataLayer reconstructs the object through the - # vocabulary registry, returning the wire-format class that - # round-trips correctly. - status_from_dl = self.datalayer.read(status_id) - if status_from_dl is None: - self.logger.warning( - "%s: status '%s' not readable from DataLayer after" - " save — skipping participant update", - self.name, - status_id, - ) - return Status.SUCCESS - - participant.participant_statuses.append( - cast(ParticipantStatus, status_from_dl) - ) - self.datalayer.save(participant) - - self.logger.info( - "%s: applied ledger status update '%s' to participant '%s'" - " (DEMOMA-07-003 step 3 receiver-side)", - self.name, - status_id, - participant_id, - ) - return Status.SUCCESS - - class ApplyNoteFromLedgerNode(DataLayerAction): """Apply an ``add_note_to_case`` ledger entry to the local case replica. diff --git a/vultron/core/behaviors/sync/nodes/participant_status_effect.py b/vultron/core/behaviors/sync/nodes/participant_status_effect.py new file mode 100644 index 000000000..c7976e5a3 --- /dev/null +++ b/vultron/core/behaviors/sync/nodes/participant_status_effect.py @@ -0,0 +1,282 @@ +#!/usr/bin/env python + +# Copyright (c) 2026 Carnegie Mellon University and Contributors. +# - see Contributors.md for a full list of Contributors +# - see ContributionInstructions.md for information on how you can Contribute to this project +# Vultron Multiparty Coordinated Vulnerability Disclosure Protocol Prototype is +# licensed under a MIT (SEI)-style license, please see LICENSE.md distributed +# with this Software or contact permission@sei.cmu.edu for full terms. +# Created, in part, with funding and support from the United States Government +# (see Acknowledgments file). This program may include and/or can make use of +# certain third party source code, object code, documentation and other files +# ("Third Party Software"). See LICENSE.md for more details. +# Carnegie Mellon®, CERT® and CERT Coordination Center® are registered in the +# U.S. Patent and Trademark Office by Carnegie Mellon University + +"""Ledger effect node for add_participant_status_to_participant entries. + +Provides :class:`ApplyParticipantStatusFromLedgerNode`, which applies an +``add_participant_status_to_participant`` ledger entry to the local participant +record, and the RM ratchet that keeps that application monotonic (RSH-05-007, +ADR-0061). + +Per specs/multi-actor-demo.yaml DEMOMA-07-003 step 3 and +specs/sync-ledger-replication.yaml SYNC-02-002. +""" + +from __future__ import annotations + +import logging +from typing import Any, cast + +import py_trees +from py_trees.common import Status + +from vultron.core.behaviors.helpers import DataLayerAction, read_rm_states +from vultron.core.behaviors.sync.nodes.effects import _extract_id_from_field +from vultron.core.models._helpers import _as_id +from vultron.core.models.case_participant import CaseParticipant +from vultron.core.models.dimensions import RmDimension +from vultron.core.models.participant_status import ParticipantStatus +from vultron.core.states.rm import RM, is_rm_at_least + +logger = logging.getLogger(__name__) + + +def _ratchet_rm( + status_obj: ParticipantStatus, local_rm: RM | None +) -> tuple[ParticipantStatus, RM | None]: + """Carry *local_rm* forward when *status_obj* would regress it. + + Monotonic visibility (``notes/sync-ledger-replication.md``): a replica must + never move an RM state backwards on the progress scale, even on an entry + from the authoritative Case Actor. A replayed, reordered, or divergent + entry would otherwise un-see progress the replica has already observed. + + Lateral moves at the same rank (``VALID`` ↔ ``INVALID``, + ``DEFERRED`` ↔ ``ACCEPTED``) are *not* regressions: the Case Actor is + authoritative for re-adjudication and those are applied unchanged. + + Returns: + The status to record and the refused RM value, or ``(status_obj, None)`` + when nothing was refused. + """ + if local_rm is None: + return status_obj, None + entry_rm = status_obj.rm.state + if entry_rm == local_rm or is_rm_at_least(entry_rm, local_rm): + return status_obj, None + return ( + status_obj.model_copy( + update={"rm": RmDimension(state=local_rm), "name": None} + ), + entry_rm, + ) + + +class ApplyParticipantStatusFromLedgerNode(DataLayerAction): + """Apply an ``add_participant_status_to_participant`` ledger entry locally. + + When a non-Case-Actor participant receives + ``Announce(CaseLedgerEntry)`` and the entry's ``event_type`` is + ``add_participant_status_to_participant``, this node reconstructs the + :class:`~vultron.core.models.participant_status.ParticipantStatus` from + the entry's ``payload_snapshot`` and appends it to the matching + :class:`~vultron.core.models.case_participant.CaseParticipant` in the + local DataLayer. + + The Case Actor is considered authoritative for *which* transition happened + — it already adjudicated the assertion before committing the entry — so + this node does not re-run the RM transition rules. It does enforce one + invariant the Case Actor cannot vouch for from the replica's vantage point: + RM state must never move backwards on the progress scale (monotonic + visibility). A replayed, reordered, or divergent entry that would regress + the local RM state has that dimension carried forward at the local value; + every other dimension is applied as the entry describes it. Lateral moves + at the same rank (``VALID`` ↔ ``INVALID``) are applied unchanged. + + Idempotency is preserved — if the status ID is already present in the + participant's list, the node returns SUCCESS without modifying the + DataLayer. + + Lenient on missing data: if the participant is not found in the local + DataLayer (this actor may have a partial view of the case), or the + payload snapshot is incomplete, the node returns SUCCESS without error to + avoid blocking the ``Announce`` processing flow. It is *not* lenient on a + malformed local record: a participant whose recorded status is not + core-shaped yields FAILURE, because the ratchet cannot be enforced against + an unreadable floor (ARCH-15-001, ADR-0062). + + Per specs/multi-actor-demo.yaml DEMOMA-07-003 step 3, + specs/sync-ledger-replication.yaml SYNC-02-002. + """ + + def setup(self, **kwargs: Any) -> None: + super().setup(**kwargs) + self.blackboard.register_key( + key="activity", access=py_trees.common.Access.READ + ) + + def _apply_rm_ratchet( + self, + status_obj: ParticipantStatus, + participant: CaseParticipant, + status_id: str, + participant_id: str, + ) -> ParticipantStatus | None: + """Enforce monotonic RM visibility, logging a carried-forward value. + + Returns ``None`` when the participant's recorded status is not + core-shaped, and the caller must then return ``Status.FAILURE``. A + shape mismatch is not an absence: reading it as "no local RM known" + would hand :func:`_ratchet_rm` a ``None`` floor and skip the ratchet + entirely, letting a regressing entry through unchecked — the defect + behind #2264 (ARCH-15-001, ARCH-15-002, ADR-0062). Genuine absence — + a replica whose participant record carries no status yet — has no + floor to enforce and is handled here as such. + """ + current = getattr(participant, "participant_status", None) + local_rm: RM | None = None + if current is not None: + states = read_rm_states(self, current) + if states is None: + return None + (local_rm,) = states + + ratcheted, refused_rm = _ratchet_rm(status_obj, local_rm) + if refused_rm is not None: + self.logger.warning( + "%s: ledger entry for '%s' would regress participant '%s'" + " from rm=%s to rm=%s — carrying the local value forward" + " (monotonic visibility, SYNC-02-002)", + self.name, + status_id, + participant_id, + ratcheted.rm.state, + refused_rm, + ) + return ratcheted + + def update(self) -> Status: + if (f := self._require_datalayer()) is not None: + return f + assert self.datalayer is not None + from vultron.core.behaviors.sync.nodes.conditions import ( + _require_log_entry, + ) + + entry = _require_log_entry(self.blackboard.activity, self.name) + snapshot = entry.payload_snapshot + + status_data = snapshot.get("object") + target_data = snapshot.get("target") + + status_id = _extract_id_from_field(status_data) + participant_id = _extract_id_from_field(target_data) + + if not status_id or not participant_id: + self.logger.debug( + "%s: payload_snapshot missing 'object' or 'target' id" + " — skipping status apply (non-fatal)", + self.name, + ) + return Status.SUCCESS + + participant = self.datalayer.read(participant_id) + if not isinstance(participant, CaseParticipant): + self.logger.debug( + "%s: participant '%s' not found in local DataLayer" + " — skipping (non-fatal, partial case view)", + self.name, + participant_id, + ) + return Status.SUCCESS + + existing_ids = [_as_id(s) for s in participant.participant_statuses] + if status_id in existing_ids: + self.logger.debug( + "%s: status '%s' already present on participant '%s'" + " — idempotent no-op", + self.name, + status_id, + participant_id, + ) + return Status.SUCCESS + + if not isinstance(status_data, dict): + self.logger.warning( + "%s: payload_snapshot 'object' is not a dict" + " — cannot reconstruct ParticipantStatus for '%s'", + self.name, + status_id, + ) + return Status.SUCCESS + + try: + status_obj = ParticipantStatus.model_validate(status_data) + except Exception as exc: + self.logger.warning( + "%s: failed to reconstruct ParticipantStatus from" + " payload_snapshot for '%s': %s", + self.name, + status_id, + exc, + ) + return Status.SUCCESS + + ratcheted = self._apply_rm_ratchet( + status_obj, participant, status_id, participant_id + ) + if ratcheted is None: + self.logger.error( + "%s: cannot enforce monotonic RM visibility for participant" + " '%s' — its recorded status is not core-shaped; refusing to" + " apply ledger entry '%s' (ARCH-15-001, ADR-0062)", + self.name, + participant_id, + status_id, + ) + return Status.FAILURE + status_obj = ratcheted + + # Saved unconditionally: the read-back below is what actually reaches + # ``participant_statuses``, so skipping the save when the object already + # exists locally would silently discard the RM ratchet applied above and + # append the un-ratcheted status instead — regressing the replica's RM + # while the ratchet's own log line claims the opposite (RSH-05-007, + # SYNC-02-002). + self.datalayer.save(status_obj) + + # Read back from the DataLayer to obtain the vocabulary-typed + # (wire-format) version of the status object. Appending the + # core-model instance directly to ``participant_statuses`` + # (typed ``list[WireParticipantStatus]``) causes Pydantic to + # serialize the list with default field values instead of the + # actual values, because the declared element type governs + # serialization when the runtime type differs. Reading back + # via the DataLayer reconstructs the object through the + # vocabulary registry, returning the wire-format class that + # round-trips correctly. + status_from_dl = self.datalayer.read(status_id) + if status_from_dl is None: + self.logger.warning( + "%s: status '%s' not readable from DataLayer after" + " save — skipping participant update", + self.name, + status_id, + ) + return Status.SUCCESS + + participant.participant_statuses.append( + cast(ParticipantStatus, status_from_dl) + ) + self.datalayer.save(participant) + + self.logger.info( + "%s: applied ledger status update '%s' to participant '%s'" + " (DEMOMA-07-003 step 3 receiver-side)", + self.name, + status_id, + participant_id, + ) + return Status.SUCCESS diff --git a/vultron/core/states/cs.py b/vultron/core/states/cs.py index 2b58faa1e..b66f026d1 100644 --- a/vultron/core/states/cs.py +++ b/vultron/core/states/cs.py @@ -613,6 +613,81 @@ def is_valid_pxa_transition(source: CS_pxa, dest: CS_pxa) -> bool: ) +def _is_component_regression( + source_component: str, dest_component: str +) -> bool: + """Return True if a single V/F/D or P/X/A component un-sets itself. + + Each component is a two-valued flag whose lowercase form means "has not + happened yet" and whose uppercase form means "has happened" (e.g. + ``VendorAwareness.VENDOR_UNAWARE = "v"`` vs ``VENDOR_AWARE = "V"``). + Every one of these facts is a one-way latch: once a vendor is aware, a fix + is ready, an exploit is public, or attacks are observed, that cannot become + untrue. A component therefore regresses exactly when it goes from + uppercase to lowercase. + """ + return str(source_component).isupper() and str(dest_component).islower() + + +def _is_monotonic_forward( + source: VfdState | PxaState, dest: VfdState | PxaState +) -> bool: + """Return True if *dest* strictly advances *source* with no component + regressing. + + ``source`` and ``dest`` are the ``NamedTuple`` values of a ``CS_vfd`` or + ``CS_pxa`` member; their components are compared position-wise. + """ + if source == dest: + return False + return not any( + _is_component_regression(s, d) for s, d in zip(source, dest) + ) + + +def is_monotonic_vfd_forward(source: CS_vfd, dest: CS_vfd) -> bool: + """Return True if (source → dest) advances VFD without regressing. + + ``is_valid_vfd_transition`` only recognises the three *adjacent* + single-component steps of the VFD machine (``vfd → Vfd → VFd → VFD``). + A peer may legitimately report a state several steps ahead — e.g. a vendor + that became aware, readied and deployed a fix between two status updates + reports ``vfd → VFD`` in one message. That is monotone but not adjacent, + so it needs this weaker check. + + Equality returns ``False`` (nothing advanced); callers that treat a status + confirmation as acceptable must test equality separately. Mirrors + :func:`vultron.core.states.rm.is_monotonic_rm_forward`. + + Examples:: + + is_monotonic_vfd_forward(CS_vfd.vfd, CS_vfd.VFD) # True + is_monotonic_vfd_forward(CS_vfd.Vfd, CS_vfd.Vfd) # False (no change) + is_monotonic_vfd_forward(CS_vfd.VFd, CS_vfd.Vfd) # False (F un-set) + """ + return _is_monotonic_forward(source.value, dest.value) + + +def is_monotonic_pxa_forward(source: CS_pxa, dest: CS_pxa) -> bool: + """Return True if (source → dest) advances PXA without regressing. + + The P/X/A components are mutually independent one-way latches, so any + combination of them being newly set is monotone forward — including + multi-component jumps such as ``pxa → PXA`` that + :func:`is_valid_pxa_transition` does not recognise. + + Equality returns ``False`` (nothing advanced). Mirrors + :func:`vultron.core.states.rm.is_monotonic_rm_forward`. + + Examples:: + + is_monotonic_pxa_forward(CS_pxa.pxa, CS_pxa.PXa) # True + is_monotonic_pxa_forward(CS_pxa.Pxa, CS_pxa.pxa) # False (P un-set) + is_monotonic_pxa_forward(CS_pxa.PxA, CS_pxa.PXa) # False (A un-set) + """ + return _is_monotonic_forward(source.value, dest.value) + + def create_vfd_machine() -> Machine: """ Generates a new Case State Vendor Fix Deploy Machine object