feat(layer4): live messages-table wiring for proximity triggers

Finishes the steer/transmit loop — advisories now reach the agents' sessions.

- message-sink.js: createEccMessageSink() delivers via the canonical writer
  'ecc-tui messages send' (maps steer/hold -> conflict kind, transmit -> query),
  resolving the binary from override/env/built target/PATH. Injectable runner;
  best-effort (a missing binary/failed send is counted skipped, never blocks).
- proximity.js: createProximityDispatcher() adds per-trigger cooldown so a
  persistent collision fires once then stays quiet (agents get steered, not
  spammed); runProximityTick() builds the snapshot and dispatches.
- scripts/proximity-tick.js: thin CLI — one-shot, --dry-run, --watch <sec>.
  Messages are internal ECC agent-to-agent coordination, not any external channel.
- 14 new tests (sink argv/kind mapping, cooldown dedup, tick dispatch/dry-run,
  CLI parse). Full suite 2891/2891; lint green.
This commit is contained in:
Affaan Mustafa
2026-06-20 20:49:17 -04:00
parent bd1be0c1ce
commit 71d22d0a77
5 changed files with 397 additions and 2 deletions
+70
View File
@@ -0,0 +1,70 @@
'use strict';
/**
* Concrete message sink for proximity triggers: delivers a session-to-session
* message through the canonical writer, the `ecc-tui messages send` CLI. The CLI
* owns the ecc2 session DB (the `messages` table the control pane reads), so we
* shell out to it rather than writing the SQLite directly and racing the daemon.
*
* Best-effort: if the binary is not found / the command fails, the call throws,
* and the dispatcher counts it as skipped — proximity never blocks on delivery.
* The command runner and binary path are injectable for tests.
*/
const fs = require('fs');
const path = require('path');
const { execFileSync } = require('child_process');
// Proximity trigger type → ecc-tui message kind (value_enum on `--kind`).
// A steer/hold is a collision warning; a transmit is a "what are you doing" query.
const KIND_BY_TYPE = {
proximity_steer: 'conflict',
proximity_hold: 'conflict',
proximity_transmit: 'query'
};
/**
* Resolve the ecc-tui binary: explicit override, env var, a built target in the
* repo, then the bare name (hope it's on PATH).
*/
function resolveEccBin(deps = {}) {
if (deps.binPath) return deps.binPath;
if (process.env.ECC_TUI_BIN && process.env.ECC_TUI_BIN.trim()) return process.env.ECC_TUI_BIN.trim();
const repoRoot = deps.repoRoot || path.join(__dirname, '..', '..', '..');
for (const rel of ['ecc2/target/release/ecc-tui', 'ecc2/target/debug/ecc-tui']) {
const candidate = path.join(repoRoot, rel);
try {
if (fs.existsSync(candidate)) return candidate;
} catch {
/* ignore */
}
}
return 'ecc-tui';
}
/**
* Build the `messages send` argv for a proximity message.
*/
function buildSendArgs({ fromSession, toSession, content, msgType }) {
const kind = KIND_BY_TYPE[msgType] || 'query';
return ['messages', 'send', '--from', String(fromSession), '--to', String(toSession), '--kind', kind, '--text', String(content)];
}
/**
* Create a `sendMessage({ fromSession, toSession, content, msgType })` sink that
* delivers via `ecc-tui messages send`. Inject `runCommand(bin, args)` for tests.
*/
function createEccMessageSink(deps = {}) {
const run = deps.runCommand || ((bin, args) => execFileSync(bin, args, { encoding: 'utf8', timeout: 5000, stdio: ['ignore', 'pipe', 'pipe'] }));
const bin = resolveEccBin(deps);
return function sendMessage(message) {
run(bin, buildSendArgs(message));
};
}
module.exports = {
KIND_BY_TYPE,
resolveEccBin,
buildSendArgs,
createEccMessageSink
};
+74 -1
View File
@@ -179,11 +179,84 @@ function dispatchProximityTriggers(triggers, deps = {}) {
return { dispatched, skipped };
}
/**
* Stateful dispatcher with per-trigger cooldown, so a collision that persists
* across many ticks fires once and then stays quiet until it clears or the
* cooldown lapses — agents get steered, not spammed. Inject `sendMessage`
* (e.g. createEccMessageSink) and optionally `now`/`cooldownMs` for tests.
*/
function createProximityDispatcher(deps = {}) {
const send = deps.sendMessage;
const cooldownMs = Number.isFinite(deps.cooldownMs) ? deps.cooldownMs : 5 * 60 * 1000;
const now = typeof deps.now === 'function' ? deps.now : () => Date.now();
const lastFired = new Map();
const keyOf = t => `${t.to}<-${t.from}:${t.type}`;
return {
dispatch(triggers) {
let dispatched = 0;
let suppressed = 0;
let skipped = 0;
for (const t of triggers || []) {
const key = keyOf(t);
const last = lastFired.get(key);
const ts = now();
if (last !== undefined && ts - last < cooldownMs) {
suppressed += 1;
continue;
}
if (typeof send !== 'function') {
skipped += 1;
continue;
}
try {
send({ fromSession: t.from, toSession: t.to, content: t.content, msgType: t.type });
lastFired.set(key, ts);
dispatched += 1;
} catch {
skipped += 1;
}
}
return { dispatched, suppressed, skipped };
},
reset() {
lastFired.clear();
}
};
}
/**
* One proximity tick: build the snapshot, then dispatch its triggers (steer the
* agents). `buildSnapshot()` returns a control-pane snapshot with a `proximity`
* field; `dispatcher` is a createProximityDispatcher. `dryRun` reports what
* would fire without sending. Both are injected so the CLI stays a thin wrapper
* and the logic is unit-testable.
*/
async function runProximityTick(deps = {}) {
const snapshot = await deps.buildSnapshot();
const prox = (snapshot && snapshot.proximity) || { advisories: [], triggers: [], counts: {} };
const triggers = prox.triggers || [];
let result;
if (deps.dryRun || !deps.dispatcher) {
result = { dispatched: 0, suppressed: 0, skipped: triggers.length, dryRun: Boolean(deps.dryRun) };
} else {
result = deps.dispatcher.dispatch(triggers);
}
return {
counts: prox.counts || {},
advisories: prox.advisories || [],
triggers,
result
};
}
module.exports = {
buildProximitySnapshot,
sessionsToAgents,
defaultWorkingSetFor,
defaultChangedFilesFor,
parseDiffRanges,
dispatchProximityTriggers
dispatchProximityTriggers,
createProximityDispatcher,
runProximityTick
};
+111
View File
@@ -0,0 +1,111 @@
#!/usr/bin/env node
'use strict';
/**
* Proximity tick — the live loop that turns the agent-space distance metric into
* action: scan the airspace from the control-pane state, then steer/transmit by
* sending session-to-session messages via `ecc-tui messages send`.
*
* node scripts/proximity-tick.js # one shot, deliver triggers
* node scripts/proximity-tick.js --dry-run # show what would fire, send nothing
* node scripts/proximity-tick.js --watch 30 # re-scan every 30s (dedupes per cooldown)
*
* Messages are internal ECC agent-to-agent coordination (the ecc2 `messages`
* table) — not any external channel.
*/
const { resolveControlPaneConfig, buildControlPaneSnapshot } = require('./lib/control-pane/state');
const { createProximityDispatcher, runProximityTick } = require('./lib/control-pane/proximity');
const { createEccMessageSink } = require('./lib/control-pane/message-sink');
function parseArgs(argv) {
const args = argv.slice(2);
const parsed = { watchSec: 0, dryRun: false, json: false, help: false, dbPath: null, stateDbPath: null };
for (let i = 0; i < args.length; i += 1) {
const a = args[i];
if (a === '--help' || a === '-h') parsed.help = true;
else if (a === '--dry-run') parsed.dryRun = true;
else if (a === '--json') parsed.json = true;
else if (a === '--watch') {
const v = Number.parseInt(args[i + 1], 10);
if (!Number.isInteger(v) || v <= 0) throw new Error('--watch needs a positive seconds value');
parsed.watchSec = v;
i += 1;
} else if (a === '--db') {
parsed.dbPath = args[i + 1];
i += 1;
} else if (a === '--state-db') {
parsed.stateDbPath = args[i + 1];
i += 1;
} else {
throw new Error(`Unknown argument: ${a}`);
}
}
return parsed;
}
function showHelp() {
console.log(
[
'Usage: node scripts/proximity-tick.js [--watch <seconds>] [--dry-run] [--json] [--db <path>] [--state-db <path>]',
'',
'Scan agent proximity from the control-pane state and steer/transmit by sending',
'internal session-to-session messages. --dry-run sends nothing; --watch loops.'
].join('\n')
);
}
function report(tick, json) {
if (json) {
console.log(JSON.stringify(tick, null, 2));
return;
}
const c = tick.counts || {};
console.log(
`proximity: ${c.agents ?? 0} agents, ${c.advisories ?? 0} advisories (${c.resolutions ?? 0} resolutions) — ` +
`dispatched ${tick.result.dispatched}, suppressed ${tick.result.suppressed}, skipped ${tick.result.skipped}` +
(tick.result.dryRun ? ' [dry-run]' : '')
);
for (const adv of tick.advisories.slice(0, 8)) {
const who = adv.level === 'resolution' ? `${adv.steer} steers (yields to ${adv.hold})` : 'both transmit intent';
console.log(` - ${(adv.risk * 100) | 0}% ${adv.level}: ${adv.aLabel || adv.a} <-> ${adv.bLabel || adv.b} => ${who}`);
}
}
async function main() {
const opts = parseArgs(process.argv);
if (opts.help) {
showHelp();
return;
}
const config = resolveControlPaneConfig(opts);
const sink = opts.dryRun ? null : createEccMessageSink({});
const dispatcher = createProximityDispatcher({ sendMessage: sink });
const buildSnapshot = () =>
buildControlPaneSnapshot({
config,
dbPath: opts.dbPath || config.dbPath,
stateDbPath: opts.stateDbPath || config.stateDbPath,
includeProximity: true
});
const once = async () => report(await runProximityTick({ buildSnapshot, dispatcher, dryRun: opts.dryRun }), opts.json);
await once();
if (opts.watchSec > 0) {
const sleep = ms => new Promise(r => setTimeout(r, ms));
for (;;) {
await sleep(opts.watchSec * 1000);
await once();
}
}
}
if (require.main === module) {
main().catch(err => {
process.stderr.write(`proximity-tick error: ${err.message}\n`);
process.exit(1);
});
}
module.exports = { parseArgs };