# T1.6 — `<triggergroup>` Declarative Join — Design & Implementation Plan

**Status:** Design locked + **GREEN-LIT (2026-07-24)** — full surface this sprint. Implementation starting.
**Author:** Jarvis, from analysis with Thor (2026-07-23); scope confirmed 2026-07-24.
**Repo:** `~/c/partner/psyclone2` (branch `master`).

### Scope decision (Thor, 2026-07-24)
- **Full locked-design surface in v1** (not the minimal all-present-first path) — "we need the full path soon anyway." Supersedes the older `T1-implementation-plan.md` §T1.6 open items U–X (which leaned minimal). Where the two plans conflict, **this doc wins**.
- **`within=` kept, `post=` dropped** (both meant post-time window).
- **`timeout` attr shape confirmed:** `timeout="<ms>"` + `ontimeout="fire|discard"` (default `fire` = partial-fire + L1 warn).

### Ground-truth confirmed in code (2026-07-24)
- `TriggerSpec` (`Subscriptions.h:447`) is a **packed, size-prefixed struct with an appended tail** (Filter/Retrieve/Query/Post/Signal specs live after `sizeof(TriggerSpec)`, reached by byte-offset; see `getFilterSpec` etc.). ⇒ **`TriggerGroupSpec` must follow the same packed-tail discipline** — member `TriggerSpec`s appended in the tail, offset-accessed. **No `std::vector`/heap pointers in the on-wire struct** (it syncs via shared memory / CMSDK networking; pointers would corrupt cross-node, cf. the ODR/`make clean` hazard).
- `grouppolicy`/`groupparam`/`group` (`Subscriptions.h:523–526`) are documented **load-balancing** fields ("distributing triggers within a group of components"). Confirmed **do NOT reuse** — new struct.
- Delivery path confirmed: `Node::deliverTriggerMessage` (`Node.cpp:2612`) wraps each match as a `CTRL_TRIGGER` DataMessage (`setData("TriggerSpec")` + `setAttachedMessage("Message")`), honours `spec->delay` via `delayedLocalMsgs`/`delayedNetworkMsgs`, then `addToMsgQ(procID,...)` locally or `sendMessageToNode` remotely. **Join buffer hooks in here, before the enqueue** — reuse `delayedLocalMsgs`/`MessageTimer` for maxage/timeout sweeps.
- Benchmark harness lives in the **workspace, not the repo**: `projects/cmlabs-website/benchmark-data/differentiators/` (`specs/diff_join.xml` + `src/DiffBench.cpp` → `DiffBench_JoinDriver/JoinSink/JoinCounter`). Today JoinSink hand-rolls a per-round bitmask (posts `diff.joined` only at mask=7). **G-B1 = rewrite that join declaratively** with `<triggergroup>` and prove wakeups ≈ joins.

---

## 1. Motivation & user story

Today a module is triggered by **one** subscribed message at a time; the scheduler
wakes the crank once per matching message. A module that needs *several* related
messages before it can act must gate them itself with crank-side state (a bitmask).

`<triggergroup>` makes that a **declarative join operator**. An input data event is
tagged with a unique id; as processing cascades through multiple paths, downstream
messages carry (or inherit) that tag. A module subscribes to a *group* of triggers
and is activated **once**, with **all** member messages of a completed set delivered
together, keyed by trigger name.

This is the core primitive for **multi-modal fusion** (audio + text + vision + sensor
state joined before an LLM/decision crank runs).

---

## 2. Current state (ground truth from the code)

| Piece | Reality today |
|---|---|
| `<triggergroup>` parser | `Node.cpp` (~L4257) is a **no-op** (`// register types`). Does **not** parse child `<trigger>`s into a group. |
| `TriggerSpec.grouppolicy/groupparam/group` | These are for **distributing triggers across component instances** (load-balancing), **not** join semantics. **Do not reuse** — add a new group struct. |
| Delivery | One-at-a-time. `PsyAPI::inputQueue` = `std::queue<TriggerAndMessage>` (`pair<DataMessage* trigger, DataMessage* msg>`). `waitForNewMessage(ms, triggerName)` pops one. |
| Tagging | Every `DataMessage` carries a single `uint32 tag` (`getTag/setTag`). Tag **names** sync across nodes via the `TagID` mechanism (`Node::syncTagID`). Trigger `tag=` already parsed; `tag="*"` ⇒ module-name tag. |
| Time fields | `getCreatedTime()` = **content time** (`time=`), `getSendTime()` = **post time** (`within=`). |
| Delivery path | `Node::deliverTriggerMessage()` wraps each match as a `CTRL_TRIGGER` message → per-component queue → `addInputTrigger` → `inputQueue`. |

**Design consequence:** the join buffer must sit **before** the per-component enqueue,
on the **Node side**, so wakeups ≈ joins (not one per member) and network-arriving
members are handled uniformly.

---

## 3. XML surface (final)

```xml
<module name="Simple" node="Node0">
  <triggergroup name="ts" count="2" maxage="2000">
    <trigger name="t1" type="bla1" />
    <trigger name="t2" type="bla2" />
    <trigger name="t3" type="bla3" optional="true" maxage="200" />
  </triggergroup>
  <crank name="c1" function="lib::function" />
  <post name="p1" type="bla4" to="WB/Stream/Cat" />
</module>
```

### `<triggergroup>` attributes

| Attr | Type | Meaning |
|---|---|---|
| `name` | string | **Required, unique within module.** Duplicate name ⇒ discard the extra group + `LogPrint` **level 1** warning. |
| `maxage` | ms | **Required, no default.** Whole partial set is evicted if it lives longer than this. |
| `count` | int / `"all"` | Fire when **any** N members present (`"all"` = every named member). **Default = `all`.** For "N of M", any N complete the set; members arriving after the set closes are **discarded**. |
| `min` | int | Fire at N members, then **re-fire** as more of the same tag-set arrive (growing/streaming set). |
| `within` | ms | All required present **and** their **post times** (`getSendTime`) within the window. |
| `time` | ms | All required present **and** their **content times** (`getCreatedTime`) within the window. `time="0"` = identical content time. |
| `mode` | `snapshot` | **Snapshot join:** fire on the designated **primary** trigger (first child, or `primary="true"`); attach the **latest** value of each other member — **no wait, no tag match**. |
| `slide` | bool | **Sliding-window correlation** (with `within=`): keep a rolling window, **re-emit** a fresh joined set each time a member updates while the window still holds (continuous fusion). |
| `debounce` | ms | Coalesce fast member streams — fire **at most once per N ms** with the latest set. |
| `dup` | `latest` / `keep` | On a duplicate member before the set completes: `latest` (**default**, latest-wins) or `keep` (queue all). |
| `order` | `strict` | Only fire if members arrived in **declared order** (pipeline-stage joins). Optional. |
| `timeout` | ms + policy | On window/maxage expiry with an incomplete set: **partial-fire** or **discard**. **Default = partial-fire** + `LogPrint` **level 1** warning. (Attr form TBD: `timeout="500"` + `ontimeout="fire|discard"`.) |

### `<trigger>` child attributes (additions)

| Attr | Meaning |
|---|---|
| `optional="true"` | Member is **not required** for completion; included in the delivered map if present. |
| `maxage` (per-member) | A member older than its own freshness bound **does not count** toward completion (stale vision frame vs fresh transcript). |
| `primary="true"` | Marks the primary trigger for `mode="snapshot"` (defaults to first child). |

### Tagging semantics
- Grouping key = the message **tag id**. "Same or no tag" ⇒ **untagged is a tag literally named `""`** (empty string) via `getTagID("")`; it groups on that and **never mixes** with real tags. No special-casing in the buffer.

---

## 4. SDK delivery — batched call

Existing single-message cranks are **untouched**; grouped modules opt in.

```cpp
// New batched delivery. Returns a map of trigger name -> message for the
// completed (or partial, on timeout) set. Empty map on timeout with no members.
std::map<std::string, DataMessage*> PsyAPI::waitForNewMessageGroup(
        uint32 ms, std::string &groupName);
```

- Keyed by **trigger name**; optional/absent members simply not in the map
  (crank checks `.count("t3")`).
- `order="strict"` is validated **before** building the map (a set that violates
  order is rejected/held per policy), so the map itself carries no ordering.
- Caller owns the messages for the activation lifetime, freed on next
  `waitForNewMessageGroup` (mirrors current `waitForNewMessage` ownership).

---

## 5. Join buffer (Node side)

A **join buffer** keyed by `(componentID, groupID, tagID)` accumulates member
messages. Lives in/around `Node::deliverTriggerMessage` (or a dedicated
`JoinEngine` owned by `Node`).

Per partial set, track:
- which named members have arrived (+ their messages, latest-wins or all per `dup`);
- first-arrival wall time (for group `maxage` eviction) and per-member arrival for per-member `maxage`;
- min/max `getSendTime` and `getCreatedTime` (for `within=`/`time=` gates);
- arrival order (for `order="strict"`);
- last-fire time (for `debounce=` / `min=` re-fire / `slide=`).

**Completion check** on each incoming member: all *required* present (respecting
per-member `maxage`), count/min satisfied, `within`/`time` window satisfied. On
completion → emit **one** `CTRL_TRIGGER_GROUP` activation carrying the set; close or
slide the set per policy.

**Timers:** group `maxage` and `timeout` need a scheduled sweep (reuse the existing
delayed-message / `MessageTimer` machinery) to evict/partial-fire stale sets without
waiting for the next member.

`mode="snapshot"` bypasses accumulation: on the primary member, snapshot the latest
buffered value of each other member and emit immediately.

---

## 6. Implementation steps (do NOT start yet)

### Progress
- **Steps 1–2 DONE (2026-07-24, Jarvis, direct — uncommitted for review).** First sub-agent run produced no code; implemented directly.
  - `CMSDK/include/ObjectIDs.h`: `TRIGGERGROUPDATA = GenObjID(42)`.
  - `CMSDK/include/Subscriptions.h`: new `GroupMemberSpec` (packed member: name/maxage/optional/primary) + `TriggerGroupSpec` (packed, size-prefixed, member tail, `getMemberSpec(n)`/`getMemberSpec(name)`), enums for count/dup/order/mode/ontimeout; `ComponentSpec` gains `std::list<TriggerGroupSpec*> triggerGroups` (freed via `free()` in dtor).
  - `Psyclone/src/Node.cpp` (~4257): real `<triggergroup>` parser — requires `name`+`maxage` (warn+skip if missing), dup-name discard+L1 warn, recurses each child `<trigger>` through the normal path so it routes, tags each member trigger with `group=groupID` (`getTagID(groupname)`), parses all group attrs (count/min/within/time+hasTimeWindow/debounce/timeout/mode/slide/dup/order/ontimeout) and per-member optional/maxage/primary, defaults first member primary. Stores packed group on `spec->triggerGroups`.
  - **Verified:** `g++ -fsyntax-only` on `Subscriptions.h` → EXIT 0 (only pre-existing bimap deprecation warnings). Full `make clean && make` still pending (needed before runtime tests — shared-mem struct change = ODR hazard).
  - **Decision on per-member fields:** kept in a **separate `GroupMemberSpec`** (NOT bloating the shared `TriggerSpec` binary contract). Members matched back to their registered `TriggerSpec` by name. Member triggers reuse `TriggerSpec.group` (previously an unused load-balancing field on individual triggers) purely as a back-pointer tag to the owning group id — the *join semantics* live entirely in the new `TriggerGroupSpec`, so this does not conflict with the "don't reuse grouppolicy/groupparam/group for join" rule (those remain untouched as policy fields; `group` here is just the association key).

### HANDOFF — Step 3 (JoinEngine)
- Groups are available at `spec->triggerGroups` per component; each member `TriggerSpec.group == groupSpec->id`. The grouping tag key = `groupSpec->id` (from `getTagID(name)`).
- Hook the join buffer in **`Node::deliverTriggerMessage` (`Node.cpp:2612`)**, *before* the `addToMsgQ`/`sendMessageToNode` enqueue. Key partial sets by `(componentID, groupID, msg tagID)`. On completion emit ONE `CTRL_TRIGGER_GROUP` (Step 4). Reuse `delayedLocalMsgs`/`MessageTimer` for `maxage`/`timeout` sweeps (Step 6).
- **Open for Step 3:** the group's `crankID` is not yet stamped onto `TriggerGroupSpec` (member triggers get theirs in `addTriggersToComponent`). Either stamp it there too, or resolve at fire time from a member. Recommend stamping in `addTriggersToComponent` alongside the member triggers.
- `count` sentinel: `0 = all`; `hasTimeWindow` distinguishes `time=0` from unset.

### HANDOFF — Step 3 DONE (2026-07-24, Jarvis, direct — uncommitted). Sub-agent runs (x2) wrote no usable code; implemented directly.
**Files changed (build verified: full `make all` after `make rebuild`-clean, zero errors; fresh Psyclone binary; JoinEngine symbols confirmed via `nm`):**
- `Psyclone/include/Node.h`: new `JoinEngine` class (registry + partial-set buffers + Mutex + DrumBeat sweep); `Node` gains `JoinEngine* joinEngine` + `friend class JoinEngine`.
- `Psyclone/src/Node.cpp`: (i) full `JoinEngine` impl ~269 lines (lifecycle/registry/`routeMessage`/`isComplete`/`fire`/`sweep`/`freePartial`); (ii) construct/destruct `joinEngine` in `Node`; (iii) register groups at subscription time (resolves crankID from member triggers by group id); (iv) **hot-path hook** at top of `deliverTriggerMessage`: `if (spec->group && joinEngine->isJoinGroup(...)) return joinEngine->routeMessage(...)`.
**Design:** Node-owned `JoinEngine`, guarded by `utils::Mutex` (`.enter()/.leave()`); partial sets keyed by `(groupID → tag id)`; untagged → `getTagID("")`. Single 50ms `CreateDrumBeat` sweep for maxage/timeout eviction + partial-fire. Implements dup latest/keep, within/time(+time=0), order=strict, count/min, snapshot, per-member maxage, debounce, slide-keep.
**Emit (Step-3 interim):** `fire()` synthesises ONE `CTRL_TRIGGER` activation on the primary member's message via existing `deliverTriggerMessage` (synthetic spec has group=0 → no re-entrancy) — so **wakeups ≈ joins is already demonstrable**. **Step 4 must replace this** with a real `CTRL_TRIGGER_GROUP` carrying the full member map.
**Open for Step 4/5:** (a) define `CTRL_TRIGGER_GROUP` type + `Node::deliverTriggerGroup()` packing all member msgs keyed by trigger name; (b) `PsyAPI::waitForNewMessageGroup(ms, groupName)` returning `map<string,DataMessage*>` + inputQueue support; wire `fire()` to it. (c) `removeComponent` teardown exists on JoinEngine but isn't called from a component-delete path yet (existing `messageTimer->removeComponent` also appears uncalled — wire both when the delete path is touched in T1.1).

### Remaining steps
1. ~~**Parser**~~ ✅
2. ~~**Struct**~~ ✅
3. ~~**Join buffer / JoinEngine**~~ ✅
4. ~~**CTRL_TRIGGER_GROUP + deliverTriggerGroup**~~ ✅ (2026-07-24) — type id 10307 (PsyAPI.h/.cpp + Node type registration); `Node::deliverTriggerGroup()` builds one grouped activation carrying the member map via `setAttachedMessageMap("Members", ...)` + primary `"Message"` for back-compat; `JoinEngine::fire()` now calls it (interim single-CTRL_TRIGGER emit removed); PsySpace dispatch handles CTRL_TRIGGER_GROUP like CTRL_TRIGGER.
5. ~~**SDK waitForNewMessageGroup**~~ ✅ (2026-07-24) — `PsyAPI::waitForNewMessageGroup(ms, groupName)` returns `map<string,DataMessage*>`; reuses `waitForNewMessage` dequeue, reads the `"Members"` map off the CTRL_TRIGGER_GROUP wrapper; `currentGroupMap` owned + freed on next call / in dtor. **Build: full `make all` clean, symbols confirmed via `nm`.**
   slide, debounce. Hook into `deliverTriggerMessage`.
4. **New activation type** `CTRL_TRIGGER_GROUP` + `deliverTriggerGroup(...)` that
   enqueues the whole set to the component.
5. **SDK** (`PsyAPI`): `waitForNewMessageGroup(ms, groupName)` returning the map;
   input-queue support for grouped deliveries alongside single ones.
6. **Timers:** wire group `maxage`/`timeout` sweeps into the scheduler.
7. **Tests** (`PsyTest`): count/all/min, within/time/time=0, snapshot, optional,
   per-member maxage, slide, debounce, order=strict, dup keep/latest, timeout
   partial-fire vs discard, duplicate group-name warning, empty-tag grouping.
8. **Docs (standing rule — all three in sync):** User Guide
   (`html/userguide/13-psyspec-reference.html`), System Guide
   (`html/systemguide/11-roadmap.html` → move to a shipped chapter), and the LLM
   reference (`html/psyclone-llm-guide.md`).

---

## 7. Open items for Thor — RESOLVED (2026-07-24)

- **`within=` vs `post=`:** ✅ **Collapse to `within=`, drop `post=`.**
- **`timeout` attr shape:** ✅ **`timeout="<ms>"` + `ontimeout="fire|discard"`** (default `fire`).
- **Scope:** ✅ **Full locked-design surface this sprint.**

---

## 8. Decisions locked (2026-07-23)

1. Group `maxage=` **required**, no default.
2. Untagged = tag `""`; groups on that, never mixes with real tags.
3. `dup`: latest-wins default, `keep` option.
4. `count` "N of M": **any** N; later arrivals discarded.
5. Delivery = **map** trigger-name → msg.
6. Multiple groups OK; unique names; dup ⇒ discard + L1 warn.
7. `within=` = **post time**.
8. `timeout`: partial-fire **or** discard; **default partial-fire** + L1 warn.
9. `order="strict"` supported (optional).
10. **All five** multi-modal extras in scope: snapshot, optional members,
    per-member maxage, sliding window, debounce.
