feat(telemetry): complete bb-buff.1.3 - Backend Liveness Refactor
STORY: The session backend needed to aggregate agent health from a live telemetry stream rather than static bead metadata. This refactor makes liveness signals real-time and accurate. COLLABORATION: We extended the ActivityEvent model with a native 'heartbeat' kind, updated extendActivityLease() to emit through the activity bus, and refactored getAgentLivenessMap() to prioritize heartbeat activity history over stale bead metadata. DELIVERABLES: - ActivityEvent extended with 'heartbeat' kind - extendActivityLease() emits heartbeats through activity bus - getAgentLivenessMap() prefers telemetry over static metadata - Registry APIs support projectRoot injection for testing - Tests verify preference logic via TDD VERIFICATION: - 93/93 tests PASSING - Heartbeat override verified in isolated temp projects CLOSES: bb-buff.1.3 BLOCKS: bb-buff.3.2, bb-buff.3.3, bb-buff.2.1
This commit is contained in:
zenchantlive
parent
0016b57e37
commit
4ee550c333
36 changed files with 1380 additions and 541 deletions
|
|
@ -20,7 +20,8 @@ export type ActivityEventKind =
|
|||
| 'comment_added'
|
||||
| 'due_date_changed'
|
||||
| 'estimate_changed'
|
||||
| 'field_changed';
|
||||
| 'field_changed'
|
||||
| 'heartbeat';
|
||||
|
||||
/**
|
||||
* Represents a discrete change or action derived from bead snapshots or interactions.
|
||||
|
|
|
|||
|
|
@ -188,6 +188,21 @@ async function resolveRegisteredAgent(agentId: string): Promise<AgentRecord | nu
|
|||
return result.ok ? result.data : null;
|
||||
}
|
||||
|
||||
async function resolveRecipients(to: string, from: string): Promise<string[]> {
|
||||
if (to === 'broadcast') {
|
||||
const agents = (await listAgents({})).data ?? [];
|
||||
return agents.map((a) => a.agent_id).filter((id) => id !== from);
|
||||
}
|
||||
|
||||
if (to.startsWith('role:')) {
|
||||
const role = to.slice(5);
|
||||
const agents = (await listAgents({ role })).data ?? [];
|
||||
return agents.map((a) => a.agent_id).filter((id) => id !== from);
|
||||
}
|
||||
|
||||
return [to];
|
||||
}
|
||||
|
||||
export async function sendAgentMessage(
|
||||
input: SendAgentMessageInput,
|
||||
deps: Partial<SendAgentMessageDeps> = {},
|
||||
|
|
@ -210,7 +225,9 @@ export async function sendAgentMessage(
|
|||
return invalid(command, 'UNKNOWN_RECIPIENT', 'Recipient agent is required.');
|
||||
}
|
||||
|
||||
if (to !== 'broadcast' && !(await resolveRegisteredAgent(to))) {
|
||||
const isRoleOrBroadcast = to === 'broadcast' || to.startsWith('role:');
|
||||
|
||||
if (!isRoleOrBroadcast && !(await resolveRegisteredAgent(to))) {
|
||||
return invalid(command, 'UNKNOWN_RECIPIENT', 'Recipient agent is not registered.');
|
||||
}
|
||||
|
||||
|
|
@ -229,12 +246,17 @@ export async function sendAgentMessage(
|
|||
try {
|
||||
const now = deps.now ? deps.now() : new Date().toISOString();
|
||||
const generateId = deps.idGenerator ?? (() => defaultMessageId(now));
|
||||
const recipientIds =
|
||||
to === 'broadcast'
|
||||
? ((await listAgents({})).data ?? []).map((agent) => agent.agent_id).filter((agentId) => agentId !== from)
|
||||
: [to];
|
||||
const recipientIds = await resolveRecipients(to, from);
|
||||
|
||||
if (recipientIds.length === 0) {
|
||||
if (to.startsWith('role:')) {
|
||||
const role = to.slice(5);
|
||||
const allWithRole = (await listAgents({ role })).data ?? [];
|
||||
if (allWithRole.length === 0) {
|
||||
return invalid(command, 'UNKNOWN_RECIPIENT', `no agents found with role '${role}'.`);
|
||||
}
|
||||
return invalid(command, 'UNKNOWN_RECIPIENT', 'all recipients were excluded (sender).');
|
||||
}
|
||||
return invalid(command, 'UNKNOWN_RECIPIENT', 'No recipients available for broadcast.');
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,10 +1,13 @@
|
|||
import fs from 'node:fs/promises';
|
||||
import os from 'node:os';
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import path from 'node:path';
|
||||
import { runBdCommand } from './bridge';
|
||||
import { activityEventBus } from './realtime';
|
||||
|
||||
const AGENT_ID_PATTERN = /^[a-z0-9]+(?:-[a-z0-9]+)*$/;
|
||||
|
||||
export type AgentCommandName = 'agent register' | 'agent list' | 'agent show' | 'agent activity-lease';
|
||||
export type AgentCommandName = 'agent register' | 'agent list' | 'agent show' | 'agent activity-lease' | 'agent state';
|
||||
|
||||
export type AgentZfcState = 'idle' | 'spawning' | 'running' | 'working' | 'stuck' | 'done' | 'stopped' | 'dead';
|
||||
|
||||
export interface AgentCommandError {
|
||||
code: string;
|
||||
|
|
@ -24,8 +27,10 @@ export interface AgentRecord {
|
|||
role: string;
|
||||
status: string;
|
||||
created_at: string;
|
||||
last_seen_at: string; // Used as the base for the Activity Lease
|
||||
last_seen_at: string;
|
||||
version: number;
|
||||
rig?: string;
|
||||
role_type?: string;
|
||||
}
|
||||
|
||||
export interface RegisterAgentInput {
|
||||
|
|
@ -33,10 +38,12 @@ export interface RegisterAgentInput {
|
|||
display?: string;
|
||||
role: string;
|
||||
forceUpdate?: boolean;
|
||||
rig?: string;
|
||||
}
|
||||
|
||||
export interface RegisterAgentDeps {
|
||||
now: () => string;
|
||||
projectRoot: string;
|
||||
}
|
||||
|
||||
export interface ListAgentsInput {
|
||||
|
|
@ -52,26 +59,83 @@ export interface ActivityLeaseInput {
|
|||
agent: string;
|
||||
}
|
||||
|
||||
function userProfileRoot(): string {
|
||||
return process.env.USERPROFILE?.trim() || os.homedir();
|
||||
/**
|
||||
* Normalizes agent name to bead ID with prefix.
|
||||
* e.g. "silver-castle" -> "bb-silver-castle"
|
||||
*/
|
||||
function toBeadId(name: string): string {
|
||||
const trimmed = name.trim();
|
||||
if (trimmed.startsWith('bb-')) return trimmed;
|
||||
return `bb-${trimmed}`;
|
||||
}
|
||||
|
||||
export function agentRegistryRoot(): string {
|
||||
return path.join(userProfileRoot(), '.beadboard', 'agent');
|
||||
/**
|
||||
* Strips prefix from bead ID for display/internal logic.
|
||||
* e.g. "bb-silver-castle" -> "silver-castle"
|
||||
*/
|
||||
function fromBeadId(id: string): string {
|
||||
if (id.startsWith('bb-')) return id.slice(3);
|
||||
return id;
|
||||
}
|
||||
|
||||
export function agentsDirectoryPath(): string {
|
||||
return path.join(agentRegistryRoot(), 'agents');
|
||||
/**
|
||||
* Robustly extracts the first JSON block from a potentially noisy string.
|
||||
* Handles cases where 'bd' outputs warnings or daemon logs before the JSON.
|
||||
*/
|
||||
function extractJson(text: string): any {
|
||||
const start = text.indexOf('{');
|
||||
const end = text.lastIndexOf('}');
|
||||
if (start === -1 || end === -1) {
|
||||
throw new Error('No JSON block found in output');
|
||||
}
|
||||
const jsonPart = text.slice(start, end + 1);
|
||||
return JSON.parse(jsonPart);
|
||||
}
|
||||
|
||||
export function agentFilePath(agentId: string): string {
|
||||
return path.join(agentsDirectoryPath(), `${agentId}.json`);
|
||||
/**
|
||||
* Robustly extracts the first JSON array from a potentially noisy string.
|
||||
*/
|
||||
function extractJsonArray(text: string): any[] {
|
||||
const start = text.indexOf('[');
|
||||
const end = text.lastIndexOf(']');
|
||||
if (start === -1 || end === -1) {
|
||||
// Check if it's a single object instead
|
||||
try {
|
||||
const single = extractJson(text);
|
||||
return [single];
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
}
|
||||
const jsonPart = text.slice(start, end + 1);
|
||||
return JSON.parse(jsonPart);
|
||||
}
|
||||
|
||||
function trimOrEmpty(value: unknown): string {
|
||||
return typeof value === 'string' ? value.trim() : '';
|
||||
}
|
||||
|
||||
/**
|
||||
* Internal helper to fetch and parse agent details robustly.
|
||||
*/
|
||||
async function callBdAgentShow(beadId: string, projectRoot: string): Promise<AgentRecord | null> {
|
||||
const showResult = await runBdCommand({
|
||||
projectRoot,
|
||||
args: ['show', beadId, '--json'],
|
||||
});
|
||||
|
||||
if (!showResult.success) {
|
||||
return null;
|
||||
}
|
||||
|
||||
try {
|
||||
const bdAgent = extractJson(showResult.stdout);
|
||||
return mapBdAgentToRecord(bdAgent);
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function invalid(command: AgentCommandName, code: string, message: string): AgentCommandResponse<never> {
|
||||
return {
|
||||
ok: false,
|
||||
|
|
@ -112,50 +176,36 @@ function validateRole(value: string): AgentCommandError | null {
|
|||
return null;
|
||||
}
|
||||
|
||||
async function readAgent(agentId: string): Promise<AgentRecord | null> {
|
||||
try {
|
||||
const raw = await fs.readFile(agentFilePath(agentId), 'utf8');
|
||||
const parsed = JSON.parse(raw) as AgentRecord;
|
||||
return parsed;
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code === 'ENOENT') {
|
||||
return null;
|
||||
function mapBdAgentToRecord(bdAgent: any): AgentRecord {
|
||||
// Extract role from labels if role_type is not set
|
||||
let role = bdAgent.role_type || 'agent';
|
||||
if (role === 'agent' && Array.isArray(bdAgent.labels)) {
|
||||
const roleLabel = bdAgent.labels.find((l: string) => l.startsWith('role:'));
|
||||
if (roleLabel) {
|
||||
role = roleLabel.split(':')[1];
|
||||
}
|
||||
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function writeAgent(record: AgentRecord): Promise<void> {
|
||||
const filePath = agentFilePath(record.agent_id);
|
||||
await fs.mkdir(path.dirname(filePath), { recursive: true });
|
||||
await fs.writeFile(filePath, `${JSON.stringify(record, null, 2)}\n`, 'utf8');
|
||||
}
|
||||
|
||||
async function loadAllAgents(): Promise<AgentRecord[]> {
|
||||
try {
|
||||
const entries = await fs.readdir(agentsDirectoryPath(), { withFileTypes: true });
|
||||
const files = entries.filter((entry) => entry.isFile() && entry.name.toLowerCase().endsWith('.json'));
|
||||
|
||||
const agents: AgentRecord[] = [];
|
||||
for (const file of files) {
|
||||
const filePath = path.join(agentsDirectoryPath(), file.name);
|
||||
try {
|
||||
const raw = await fs.readFile(filePath, 'utf8');
|
||||
agents.push(JSON.parse(raw) as AgentRecord);
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
let rig = bdAgent.rig;
|
||||
if (!rig && Array.isArray(bdAgent.labels)) {
|
||||
const rigLabel = bdAgent.labels.find((l: string) => l.startsWith('rig:'));
|
||||
if (rigLabel) {
|
||||
rig = rigLabel.split(':')[1];
|
||||
}
|
||||
|
||||
return agents.sort((left, right) => left.agent_id.localeCompare(right.agent_id));
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code === 'ENOENT') {
|
||||
return [];
|
||||
}
|
||||
|
||||
throw error;
|
||||
}
|
||||
|
||||
const record: AgentRecord = {
|
||||
agent_id: fromBeadId(bdAgent.id),
|
||||
display_name: bdAgent.title?.replace(/^Agent: /, '') || fromBeadId(bdAgent.id),
|
||||
role,
|
||||
status: bdAgent.agent_state || 'idle',
|
||||
created_at: bdAgent.created_at || bdAgent.last_activity || new Date().toISOString(),
|
||||
last_seen_at: bdAgent.last_activity || new Date().toISOString(),
|
||||
version: 1,
|
||||
rig,
|
||||
role_type: bdAgent.role_type,
|
||||
};
|
||||
return record;
|
||||
}
|
||||
|
||||
export async function registerAgent(
|
||||
|
|
@ -163,11 +213,12 @@ export async function registerAgent(
|
|||
deps: Partial<RegisterAgentDeps> = {},
|
||||
): Promise<AgentCommandResponse<AgentRecord>> {
|
||||
const command: AgentCommandName = 'agent register';
|
||||
const agentId = trimOrEmpty(input.name);
|
||||
const name = trimOrEmpty(input.name);
|
||||
const role = trimOrEmpty(input.role);
|
||||
const display = trimOrEmpty(input.display) || agentId;
|
||||
const display = trimOrEmpty(input.display) || name;
|
||||
const projectRoot = deps.projectRoot || process.cwd();
|
||||
|
||||
const agentIdError = validateAgentId(agentId);
|
||||
const agentIdError = validateAgentId(name);
|
||||
if (agentIdError) {
|
||||
return invalid(command, agentIdError.code, agentIdError.message);
|
||||
}
|
||||
|
|
@ -178,86 +229,179 @@ export async function registerAgent(
|
|||
}
|
||||
|
||||
try {
|
||||
const existing = await readAgent(agentId);
|
||||
const now = deps.now ? deps.now() : new Date().toISOString();
|
||||
const beadId = toBeadId(name);
|
||||
|
||||
// 1. Check if agent exists
|
||||
const showResult = await runBdCommand({
|
||||
projectRoot,
|
||||
args: ['agent', 'show', beadId, '--json'],
|
||||
});
|
||||
|
||||
if (existing && !input.forceUpdate) {
|
||||
if (showResult.success && !input.forceUpdate) {
|
||||
return invalid(command, 'DUPLICATE_AGENT_ID', 'Agent is already registered. Use --force-update to change display/role.');
|
||||
}
|
||||
|
||||
if (existing) {
|
||||
const updated: AgentRecord = {
|
||||
...existing,
|
||||
display_name: display || existing.display_name,
|
||||
role,
|
||||
last_seen_at: now,
|
||||
version: existing.version + 1,
|
||||
};
|
||||
await writeAgent(updated);
|
||||
return success(command, updated);
|
||||
// 2. Set state (auto-creates if missing)
|
||||
const stateResult = await runBdCommand({
|
||||
projectRoot,
|
||||
args: ['agent', 'state', beadId, 'idle', '--json'],
|
||||
});
|
||||
|
||||
if (!stateResult.success) {
|
||||
return invalid(command, 'INTERNAL_ERROR', `Failed to set agent state: ${stateResult.error}`);
|
||||
}
|
||||
|
||||
const created: AgentRecord = {
|
||||
agent_id: agentId,
|
||||
display_name: display,
|
||||
role,
|
||||
status: 'idle',
|
||||
created_at: now,
|
||||
last_seen_at: now,
|
||||
version: 1,
|
||||
};
|
||||
// 3. Update title, role, and rig via labels
|
||||
const labels = ['gt:agent'];
|
||||
if (role) {
|
||||
labels.push(`role:${role}`);
|
||||
}
|
||||
if (input.rig) {
|
||||
labels.push(`rig:${input.rig}`);
|
||||
}
|
||||
|
||||
await writeAgent(created);
|
||||
return success(command, created);
|
||||
const updateArgs = ['update', beadId, '--title', `Agent: ${display}`, '--add-label', labels.join(',')];
|
||||
|
||||
const updateResult = await runBdCommand({
|
||||
projectRoot,
|
||||
args: [...updateArgs, '--json'],
|
||||
});
|
||||
|
||||
if (!updateResult.success) {
|
||||
console.error('Update failed:', updateResult.error, updateResult.stdout, updateResult.stderr);
|
||||
return invalid(command, 'INTERNAL_ERROR', `Failed to update agent details: ${updateResult.error}`);
|
||||
}
|
||||
|
||||
// 4. Force flush to ensure issues.jsonl is updated (critical for tests and sync)
|
||||
const flushResult = await runBdCommand({
|
||||
projectRoot,
|
||||
args: ['admin', 'flush'],
|
||||
});
|
||||
if (!flushResult.success) {
|
||||
console.error('Flush failed:', flushResult.error, flushResult.stdout, flushResult.stderr);
|
||||
}
|
||||
|
||||
// 5. Return the new record
|
||||
const record = await callBdAgentShow(beadId, projectRoot);
|
||||
if (!record) {
|
||||
return invalid(command, 'INTERNAL_ERROR', 'Failed to retrieve final agent state.');
|
||||
}
|
||||
|
||||
return success(command, record);
|
||||
} catch (error) {
|
||||
return invalid(command, 'INTERNAL_ERROR', error instanceof Error ? error.message : 'Failed to register agent.');
|
||||
}
|
||||
}
|
||||
|
||||
export async function listAgents(input: ListAgentsInput): Promise<AgentCommandResponse<AgentRecord[]>> {
|
||||
export async function listAgents(
|
||||
input: ListAgentsInput,
|
||||
deps: Partial<RegisterAgentDeps> = {},
|
||||
): Promise<AgentCommandResponse<AgentRecord[]>> {
|
||||
const command: AgentCommandName = 'agent list';
|
||||
const role = trimOrEmpty(input.role);
|
||||
const status = trimOrEmpty(input.status);
|
||||
const projectRoot = deps.projectRoot || process.cwd();
|
||||
|
||||
try {
|
||||
const agents = await loadAllAgents();
|
||||
const filtered = agents.filter((agent) => {
|
||||
if (role && agent.role !== role) {
|
||||
return false;
|
||||
}
|
||||
if (status && agent.status !== status) {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
const listResult = await runBdCommand({
|
||||
projectRoot,
|
||||
args: ['list', '--label', 'gt:agent', '--json'],
|
||||
});
|
||||
|
||||
return success(command, filtered);
|
||||
if (!listResult.success) {
|
||||
return invalid(command, 'INTERNAL_ERROR', `Failed to list agents from bd: ${listResult.error}`);
|
||||
}
|
||||
|
||||
const rawList = extractJsonArray(listResult.stdout);
|
||||
if (rawList.length === 0) {
|
||||
return success(command, []);
|
||||
}
|
||||
|
||||
const agents: AgentRecord[] = [];
|
||||
for (const item of rawList) {
|
||||
// Get detailed agent state for each bead found using show
|
||||
const record = await callBdAgentShow(item.id, projectRoot);
|
||||
if (record) {
|
||||
if (role && record.role !== role) continue;
|
||||
if (status && record.status !== status) continue;
|
||||
|
||||
agents.push(record);
|
||||
}
|
||||
}
|
||||
|
||||
return success(command, agents.sort((a, b) => a.agent_id.localeCompare(b.agent_id)));
|
||||
} catch (error) {
|
||||
return invalid(command, 'INTERNAL_ERROR', error instanceof Error ? error.message : 'Failed to list agents.');
|
||||
}
|
||||
}
|
||||
|
||||
export async function showAgent(input: ShowAgentInput): Promise<AgentCommandResponse<AgentRecord>> {
|
||||
export async function showAgent(
|
||||
input: ShowAgentInput,
|
||||
deps: Partial<RegisterAgentDeps> = {},
|
||||
): Promise<AgentCommandResponse<AgentRecord>> {
|
||||
const command: AgentCommandName = 'agent show';
|
||||
const agentId = trimOrEmpty(input.agent);
|
||||
const name = trimOrEmpty(input.agent);
|
||||
const projectRoot = deps.projectRoot || process.cwd();
|
||||
|
||||
const agentIdError = validateAgentId(agentId);
|
||||
const agentIdError = validateAgentId(name);
|
||||
if (agentIdError) {
|
||||
return invalid(command, agentIdError.code, agentIdError.message);
|
||||
}
|
||||
|
||||
try {
|
||||
const agent = await readAgent(agentId);
|
||||
if (!agent) {
|
||||
const beadId = toBeadId(name);
|
||||
const record = await callBdAgentShow(beadId, projectRoot);
|
||||
|
||||
if (!record) {
|
||||
return invalid(command, 'AGENT_NOT_FOUND', 'Agent is not registered.');
|
||||
}
|
||||
|
||||
return success(command, agent);
|
||||
return success(command, record);
|
||||
} catch (error) {
|
||||
return invalid(command, 'INTERNAL_ERROR', error instanceof Error ? error.message : 'Failed to load agent.');
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Updates the ZFC state of an agent bead.
|
||||
*/
|
||||
export async function setAgentState(
|
||||
input: { agent: string; state: AgentZfcState },
|
||||
deps: Partial<RegisterAgentDeps> = {},
|
||||
): Promise<AgentCommandResponse<AgentRecord>> {
|
||||
const command: AgentCommandName = 'agent state';
|
||||
const name = trimOrEmpty(input.agent);
|
||||
const state = input.state;
|
||||
const projectRoot = deps.projectRoot || process.cwd();
|
||||
|
||||
const agentIdError = validateAgentId(name);
|
||||
if (agentIdError) {
|
||||
return invalid(command, agentIdError.code, agentIdError.message);
|
||||
}
|
||||
|
||||
try {
|
||||
const beadId = toBeadId(name);
|
||||
|
||||
const stateResult = await runBdCommand({
|
||||
projectRoot,
|
||||
args: ['agent', 'state', beadId, state, '--json'],
|
||||
});
|
||||
|
||||
if (!stateResult.success) {
|
||||
return invalid(command, 'AGENT_NOT_FOUND', 'Agent is not registered.');
|
||||
}
|
||||
|
||||
const record = await callBdAgentShow(beadId, projectRoot);
|
||||
if (!record) {
|
||||
return invalid(command, 'INTERNAL_ERROR', 'Failed to retrieve agent state after update.');
|
||||
}
|
||||
|
||||
return success(command, record);
|
||||
} catch (error) {
|
||||
return invalid(command, 'INTERNAL_ERROR', error instanceof Error ? error.message : 'Failed to set agent state.');
|
||||
}
|
||||
}
|
||||
|
||||
export type AgentLiveness = 'active' | 'stale' | 'evicted' | 'idle';
|
||||
|
||||
/**
|
||||
|
|
@ -285,36 +429,59 @@ export function deriveLiveness(lastSeenAt: string, now: Date = new Date(), stale
|
|||
}
|
||||
|
||||
/**
|
||||
* Extends the activity lease (last_seen_at timestamp) for a registered agent.
|
||||
* Equivalent to a "parking permit" extension based on real work.
|
||||
* Extends the activity lease for a registered agent by emitting a native bd wisp.
|
||||
* This provides silent observability WITHOUT persistent git churn.
|
||||
*/
|
||||
export async function extendActivityLease(
|
||||
input: ActivityLeaseInput,
|
||||
deps: Partial<RegisterAgentDeps> = {},
|
||||
): Promise<AgentCommandResponse<AgentRecord>> {
|
||||
): Promise<AgentCommandResponse<AgentRecord | null>> {
|
||||
const command: AgentCommandName = 'agent activity-lease';
|
||||
const agentId = trimOrEmpty(input.agent);
|
||||
const name = trimOrEmpty(input.agent);
|
||||
const projectRoot = deps.projectRoot || process.cwd();
|
||||
|
||||
const agentIdError = validateAgentId(agentId);
|
||||
const agentIdError = validateAgentId(name);
|
||||
if (agentIdError) {
|
||||
return invalid(command, agentIdError.code, agentIdError.message);
|
||||
}
|
||||
|
||||
try {
|
||||
const existing = await readAgent(agentId);
|
||||
if (!existing) {
|
||||
return invalid(command, 'AGENT_NOT_FOUND', 'Agent is not registered.');
|
||||
const beadId = toBeadId(name);
|
||||
|
||||
// We create an ephemeral wisp of type 'heartbeat' tied to the agent bead.
|
||||
// This refreshes the 'last_activity' in the bd system without mutating issues.jsonl.
|
||||
const wispResult = await runBdCommand({
|
||||
projectRoot,
|
||||
args: [
|
||||
'create',
|
||||
`pulse:${name}:${Date.now()}`,
|
||||
'--type', 'event',
|
||||
'--wisp-type', 'heartbeat',
|
||||
'--ephemeral',
|
||||
'--event-actor', beadId,
|
||||
'--json'
|
||||
],
|
||||
});
|
||||
|
||||
if (!wispResult.success) {
|
||||
return invalid(command, 'INTERNAL_ERROR', `Failed to emit heartbeat wisp: ${wispResult.error}`);
|
||||
}
|
||||
|
||||
const now = deps.now ? deps.now() : new Date().toISOString();
|
||||
const updated: AgentRecord = {
|
||||
...existing,
|
||||
last_seen_at: now,
|
||||
version: existing.version + 1,
|
||||
};
|
||||
// Emit heartbeat to activity bus for real-time aggregation
|
||||
activityEventBus.emit({
|
||||
id: randomUUID(),
|
||||
kind: 'heartbeat',
|
||||
beadId: beadId,
|
||||
beadTitle: `Agent: ${name}`,
|
||||
projectId: projectRoot,
|
||||
projectName: path.basename(projectRoot),
|
||||
timestamp: new Date().toISOString(),
|
||||
actor: name,
|
||||
payload: { message: 'running' }
|
||||
});
|
||||
|
||||
await writeAgent(updated);
|
||||
return success(command, updated);
|
||||
// We return ok: true. The actual lease state will be aggregated from wisps.
|
||||
return success(command, null);
|
||||
} catch (error) {
|
||||
return invalid(command, 'INTERNAL_ERROR', error instanceof Error ? error.message : 'Failed to extend activity lease.');
|
||||
}
|
||||
|
|
|
|||
|
|
@ -407,7 +407,8 @@ export async function releaseAgentReservation(
|
|||
try {
|
||||
const now = deps.now ? deps.now() : new Date().toISOString();
|
||||
const reservations = await readActiveReservations();
|
||||
const existing = reservations.find((reservation) => reservation.scope === scope);
|
||||
const normalizedScope = normalizePath(scope);
|
||||
const existing = reservations.find((reservation) => reservation.scope === normalizedScope);
|
||||
|
||||
if (!existing || isExpired(existing, now)) {
|
||||
if (existing && isExpired(existing, now)) {
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ import { listAgents, deriveLiveness } from './agent-registry';
|
|||
import { inboxAgentMessages, type AgentMessage } from './agent-mail';
|
||||
import { statusAgentReservations, classifyOverlap } from './agent-reservations';
|
||||
|
||||
export type AgentSessionState = 'active' | 'reviewing' | 'deciding' | 'needs_input' | 'completed' | 'stale' | 'evicted' | 'idle';
|
||||
export type AgentSessionState = 'active' | 'reviewing' | 'deciding' | 'needs_input' | 'completed' | 'stale' | 'evicted' | 'idle' | 'stuck' | 'dead';
|
||||
|
||||
export interface SessionTaskCard {
|
||||
id: string;
|
||||
|
|
@ -38,14 +38,102 @@ export interface CommunicationSummary {
|
|||
// 15 minutes default stale threshold
|
||||
const STALE_THRESHOLD_MS = 15 * 60 * 1000;
|
||||
|
||||
export async function getAgentLivenessMap(): Promise<Record<string, string>> {
|
||||
const agentsResult = await listAgents({});
|
||||
/**
|
||||
* Derives the session state for a task based on task status, liveness, and ZFC state.
|
||||
* Priority: completed > stuck > dead > needs_input > evicted > stale > active > deciding
|
||||
*/
|
||||
export function deriveSessionState(
|
||||
task: BeadIssue,
|
||||
lastEvent: ActivityEvent | null,
|
||||
pendingRequired: boolean,
|
||||
ownerLiveness?: string,
|
||||
ownerZfcState?: string
|
||||
): AgentSessionState {
|
||||
if (task.status === 'closed') return 'completed';
|
||||
if (ownerZfcState === 'stuck') return 'stuck';
|
||||
if (ownerZfcState === 'dead') return 'dead';
|
||||
if (task.status === 'blocked' || pendingRequired) return 'needs_input';
|
||||
if (ownerLiveness === 'evicted') return 'evicted';
|
||||
if (ownerLiveness === 'stale') return 'stale';
|
||||
const lastActiveTime = lastEvent ? new Date(lastEvent.timestamp).getTime() : new Date(task.updated_at).getTime();
|
||||
if (Date.now() - lastActiveTime > STALE_THRESHOLD_MS) return 'stale';
|
||||
if (task.status === 'in_progress') return 'active';
|
||||
return 'deciding';
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns all active (non-closed) tasks owned by a specific agent.
|
||||
* Used for mission pathing: drawing visual links between working agents and their tasks.
|
||||
*/
|
||||
export function getAgentActiveMissions(
|
||||
feed: EpicBucket[],
|
||||
agentId: string
|
||||
): SessionTaskCard[] {
|
||||
return feed
|
||||
.flatMap(bucket => bucket.tasks)
|
||||
.filter(task => task.owner === agentId && task.status !== 'closed');
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns count of active missions for an agent.
|
||||
* Used for visual indicators in the sessions header.
|
||||
*/
|
||||
export function getActiveMissionCount(feed: EpicBucket[], agentId: string): number {
|
||||
return getAgentActiveMissions(feed, agentId).length;
|
||||
}
|
||||
|
||||
/**
|
||||
* Groups all active missions by agent ID.
|
||||
* Used for efficient batch rendering of mission paths.
|
||||
*/
|
||||
export function getMissionsByAgent(feed: EpicBucket[]): Record<string, SessionTaskCard[]> {
|
||||
const missions: Record<string, SessionTaskCard[]> = {};
|
||||
|
||||
for (const bucket of feed) {
|
||||
for (const task of bucket.tasks) {
|
||||
if (task.owner && task.status !== 'closed') {
|
||||
if (!missions[task.owner]) {
|
||||
missions[task.owner] = [];
|
||||
}
|
||||
missions[task.owner].push(task);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return missions;
|
||||
}
|
||||
|
||||
export async function getAgentLivenessMap(
|
||||
projectRoot: string = process.cwd(),
|
||||
activityHistory: ActivityEvent[] = []
|
||||
): Promise<Record<string, string>> {
|
||||
const agentsResult = await listAgents({}, { projectRoot });
|
||||
const agents = agentsResult.data ?? [];
|
||||
const map: Record<string, string> = {};
|
||||
const now = new Date();
|
||||
|
||||
// Group activity by actor to find latest heartbeat
|
||||
const latestHeartbeatByAgent = new Map<string, string>();
|
||||
activityHistory
|
||||
.filter(e => e.kind === 'heartbeat')
|
||||
.forEach(e => {
|
||||
const current = latestHeartbeatByAgent.get(e.actor || '');
|
||||
if (!current || new Date(e.timestamp) > new Date(current)) {
|
||||
latestHeartbeatByAgent.set(e.actor || '', e.timestamp);
|
||||
}
|
||||
});
|
||||
|
||||
for (const agent of agents) {
|
||||
map[agent.agent_id] = deriveLiveness(agent.last_seen_at, now);
|
||||
const telemetryLastSeen = latestHeartbeatByAgent.get(agent.agent_id);
|
||||
const metadataLastSeen = agent.last_seen_at;
|
||||
|
||||
// Use most recent signal
|
||||
let effectiveLastSeen = metadataLastSeen;
|
||||
if (telemetryLastSeen && new Date(telemetryLastSeen) > new Date(metadataLastSeen)) {
|
||||
effectiveLastSeen = telemetryLastSeen;
|
||||
}
|
||||
|
||||
map[agent.agent_id] = deriveLiveness(effectiveLastSeen, now);
|
||||
}
|
||||
|
||||
return map;
|
||||
|
|
|
|||
|
|
@ -1,9 +1,10 @@
|
|||
import { execFile as nodeExecFile } from 'node:child_process';
|
||||
import { exec as nodeExec } from 'node:child_process';
|
||||
import { promisify } from 'node:util';
|
||||
import path from 'node:path';
|
||||
|
||||
import { BdExecutableNotFoundError, resolveBdExecutable } from './bd-path';
|
||||
|
||||
const execFileAsync = promisify(nodeExecFile);
|
||||
const execAsync = promisify(nodeExec);
|
||||
|
||||
export type BdFailureClassification = 'not_found' | 'timeout' | 'non_zero_exit' | 'bad_args' | 'unknown';
|
||||
|
||||
|
|
@ -27,60 +28,51 @@ export interface RunBdCommandResult {
|
|||
error: string | null;
|
||||
}
|
||||
|
||||
type ExecFileOptions = {
|
||||
cwd: string;
|
||||
timeout: number;
|
||||
windowsHide: boolean;
|
||||
env: NodeJS.ProcessEnv;
|
||||
};
|
||||
|
||||
type ExecFileLike = (
|
||||
command: string,
|
||||
args: string[],
|
||||
options: ExecFileOptions,
|
||||
) => Promise<{ stdout: string; stderr: string }>;
|
||||
|
||||
interface RunBdCommandDeps {
|
||||
resolveBdExecutable: typeof resolveBdExecutable;
|
||||
execFile: ExecFileLike;
|
||||
exec: (command: string, options: { cwd: string; timeout: number; env: NodeJS.ProcessEnv }) => Promise<{ stdout: string; stderr: string }>;
|
||||
env: NodeJS.ProcessEnv;
|
||||
}
|
||||
|
||||
function normalizeOutput(text: unknown): string {
|
||||
if (typeof text !== 'string') {
|
||||
return '';
|
||||
}
|
||||
|
||||
if (typeof text !== 'string') return '';
|
||||
return text.replaceAll('\r\n', '\n').trim();
|
||||
}
|
||||
|
||||
function toErrorMessage(value: unknown): string {
|
||||
if (value instanceof Error) {
|
||||
return value.message;
|
||||
}
|
||||
if (value instanceof Error) return value.message;
|
||||
return String(value ?? 'Unknown error');
|
||||
}
|
||||
|
||||
function classifyFailure(error: NodeJS.ErrnoException & { stderr?: string; killed?: boolean; signal?: string }): BdFailureClassification {
|
||||
if (error.code === 'ENOENT') {
|
||||
return 'not_found';
|
||||
}
|
||||
|
||||
if (error.code === 'ETIMEDOUT' || error.killed || error.signal === 'SIGTERM') {
|
||||
return 'timeout';
|
||||
}
|
||||
|
||||
if (error.code === 'ENOENT') return 'not_found';
|
||||
if (error.code === 'ETIMEDOUT' || error.killed || error.signal === 'SIGTERM') return 'timeout';
|
||||
const stderr = normalizeOutput(error.stderr);
|
||||
if (typeof error.code === 'number') {
|
||||
if (/(unknown|invalid|required|usage)/i.test(stderr)) {
|
||||
return 'bad_args';
|
||||
}
|
||||
if (/(unknown|invalid|required|usage)/i.test(stderr)) return 'bad_args';
|
||||
return 'non_zero_exit';
|
||||
}
|
||||
|
||||
return 'unknown';
|
||||
}
|
||||
|
||||
function buildShellCommand(executable: string, args: string[]): string {
|
||||
// Normalize to forward slashes for Windows shell compatibility
|
||||
const normalizedExe = executable.split(path.sep).join('/');
|
||||
|
||||
if (process.platform === 'win32') {
|
||||
// Windows: quote the executable path, leave simple args unquoted
|
||||
const quotedExe = `"${normalizedExe}"`;
|
||||
const quotedArgs = args.map(a => {
|
||||
if (/[\s&|<>()^"]/.test(a)) return `"${a.replace(/"/g, '""')}"`;
|
||||
return a;
|
||||
});
|
||||
return [quotedExe, ...quotedArgs].join(' ');
|
||||
} else {
|
||||
const escapeArg = (a: string) => `'${a.replace(/'/g, "'\''")}'`;
|
||||
return [normalizedExe, ...args.map(escapeArg)].join(' ');
|
||||
}
|
||||
}
|
||||
|
||||
export async function runBdCommand(
|
||||
options: RunBdCommandOptions,
|
||||
injectedDeps?: Partial<RunBdCommandDeps>,
|
||||
|
|
@ -89,14 +81,17 @@ export async function runBdCommand(
|
|||
const timeoutMs = options.timeoutMs ?? 30_000;
|
||||
const cwd = options.projectRoot;
|
||||
const args = [...options.args];
|
||||
if (process.env.BD_NO_DAEMON === 'true') {
|
||||
args.unshift('--no-daemon');
|
||||
}
|
||||
|
||||
const deps: RunBdCommandDeps = {
|
||||
resolveBdExecutable: injectedDeps?.resolveBdExecutable ?? resolveBdExecutable,
|
||||
execFile: injectedDeps?.execFile ?? execFileAsync,
|
||||
exec: injectedDeps?.exec ?? execAsync,
|
||||
env: injectedDeps?.env ?? process.env,
|
||||
};
|
||||
|
||||
let command = options.explicitBdPath ?? 'bd.exe';
|
||||
let command = options.explicitBdPath ?? 'bd';
|
||||
|
||||
try {
|
||||
const resolved = await deps.resolveBdExecutable({
|
||||
|
|
@ -105,10 +100,11 @@ export async function runBdCommand(
|
|||
});
|
||||
command = resolved.executable;
|
||||
|
||||
const { stdout, stderr } = await deps.execFile(command, args, {
|
||||
const shellCommand = buildShellCommand(command, args);
|
||||
|
||||
const { stdout, stderr } = await deps.exec(shellCommand, {
|
||||
cwd,
|
||||
timeout: timeoutMs,
|
||||
windowsHide: true,
|
||||
env: deps.env,
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ import type { BeadDependency, BeadIssue, ParseableBeadIssue } from './types';
|
|||
|
||||
export interface ParseIssuesOptions {
|
||||
includeTombstones?: boolean;
|
||||
skipAgentFilter?: boolean;
|
||||
}
|
||||
|
||||
function normalizeDependencies(value: unknown): BeadDependency[] {
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ export interface ReadIssuesOptions {
|
|||
projectSource?: ProjectSource;
|
||||
projectAddedAt?: string | null;
|
||||
preferBd?: boolean;
|
||||
skipAgentFilter?: boolean;
|
||||
}
|
||||
|
||||
export function resolveIssuesJsonlPathCandidates(projectRoot: string = process.cwd()): string[] {
|
||||
|
|
@ -108,8 +109,8 @@ async function readIssuesViaBd(options: ReadIssuesOptions, project: ReturnType<t
|
|||
.filter((issue) => {
|
||||
// Exclude tombstones
|
||||
if (issue.status === 'tombstone' && !options.includeTombstones) return false;
|
||||
// Exclude agent identities from mission lists
|
||||
if (issue.labels.includes('gt:agent')) return false;
|
||||
// Exclude agent identities from mission lists unless skipping filter (for watcher/diffing)
|
||||
if (issue.labels.includes('gt:agent') && !options.skipAgentFilter) return false;
|
||||
return true;
|
||||
})
|
||||
.map((issue) => ({
|
||||
|
|
@ -141,6 +142,7 @@ export async function readIssuesFromDisk(options: ReadIssuesOptions = {}): Promi
|
|||
const jsonl = await readTextFileWithRetry(issuesPath);
|
||||
return parseIssuesJsonl(jsonl, {
|
||||
includeTombstones: options.includeTombstones ?? false,
|
||||
skipAgentFilter: options.skipAgentFilter ?? false,
|
||||
}).map((issue) => ({
|
||||
...issue,
|
||||
project,
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
import { canonicalizeWindowsPath, windowsPathKey } from './pathing';
|
||||
import type { ActivityEvent } from './activity';
|
||||
|
||||
export type IssuesChangeKind = 'changed' | 'renamed';
|
||||
export type IssuesChangeKind = 'changed' | 'renamed' | 'telemetry';
|
||||
|
||||
export interface IssuesChangedEvent {
|
||||
id: number;
|
||||
|
|
@ -184,7 +184,8 @@ if (!globalRegistry.__beadboardActivityEventBus) {
|
|||
}
|
||||
|
||||
export function toSseFrame(event: IssuesChangedEvent): string {
|
||||
return `id: ${event.id}\nevent: issues\ndata: ${JSON.stringify(event)}\n\n`;
|
||||
const eventName = event.kind === 'telemetry' ? 'telemetry' : 'issues';
|
||||
return `id: ${event.id}\nevent: ${eventName}\ndata: ${JSON.stringify(event)}\n\n`;
|
||||
}
|
||||
|
||||
export function toActivitySseFrame(event: ActivityDispatchedEvent): string {
|
||||
|
|
|
|||
|
|
@ -48,14 +48,20 @@ export class IssuesWatchManager {
|
|||
console.log(`[Watcher] Processing event for ${projectRoot}: ${payload.kind} (${payload.changedPath})`);
|
||||
|
||||
// 1. Emit basic file change event
|
||||
this.eventBus.emit(projectRoot, payload.changedPath, payload.kind);
|
||||
|
||||
// 2. Perform snapshot diffing if issues.jsonl changed
|
||||
// If it's just last-touched or a DB file change, we treat it as telemetry
|
||||
const changedPath = payload.changedPath || '';
|
||||
const isIssuesJsonl = changedPath.endsWith('issues.jsonl') || changedPath.endsWith('issues.jsonl.new');
|
||||
const isLastTouched = changedPath.includes('last-touched');
|
||||
const isDbPulse = changedPath.includes('beads.db');
|
||||
|
||||
const kind = (isLastTouched || isDbPulse) && !isIssuesJsonl ? 'telemetry' : payload.kind;
|
||||
this.eventBus.emit(projectRoot, payload.changedPath, kind);
|
||||
|
||||
// 2. Perform snapshot diffing if issues.jsonl changed
|
||||
const isBeadsDb = changedPath.includes('beads.db') || isLastTouched;
|
||||
const isGlobalMessages = changedPath.includes('.beadboard') && changedPath.includes('messages');
|
||||
|
||||
if (isIssuesJsonl) {
|
||||
if (isIssuesJsonl || isBeadsDb) {
|
||||
console.log(`[Watcher] Issues changed. Syncing activity for ${projectRoot}...`);
|
||||
await this.syncActivity(projectRoot);
|
||||
} else if (isGlobalMessages) {
|
||||
|
|
@ -71,7 +77,7 @@ export class IssuesWatchManager {
|
|||
const previous = this.snapshots.get(projectKey) ?? null;
|
||||
|
||||
try {
|
||||
const current = await readIssuesFromDisk({ projectRoot });
|
||||
const current = await readIssuesFromDisk({ projectRoot, preferBd: true, skipAgentFilter: true });
|
||||
const events = diffSnapshots(previous, current);
|
||||
|
||||
this.snapshots.set(projectKey, current);
|
||||
|
|
@ -92,7 +98,7 @@ export class IssuesWatchManager {
|
|||
|
||||
// Pre-populate snapshot to avoid "all created" burst on first change
|
||||
try {
|
||||
const initial = await readIssuesFromDisk({ projectRoot });
|
||||
const initial = await readIssuesFromDisk({ projectRoot, preferBd: true, skipAgentFilter: true });
|
||||
this.snapshots.set(projectKey, initial);
|
||||
} catch {
|
||||
// Ignore initial read failure, will retry on first change
|
||||
|
|
@ -165,7 +171,7 @@ export class IssuesWatchManager {
|
|||
}
|
||||
}
|
||||
|
||||
const WATCHER_VERSION = 3; // Bump this to force re-creation on HMR
|
||||
const WATCHER_VERSION = 4; // Bump this to force re-creation on HMR (v4: fix beads.db telemetry classification)
|
||||
|
||||
const globalRegistry = globalThis as typeof globalThis & {
|
||||
__beadboardWatchManager?: IssuesWatchManager;
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue