mirror of
https://github.com/qwibitai/nanoclaw.git
synced 2026-06-18 18:29:35 +08:00
Compare commits
15 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 3eb4207c3f | |||
| 63a175c3df | |||
| f87dbfc75c | |||
| aad6efc874 | |||
| cebaa0246c | |||
| c1b9c43281 | |||
| b24e189af9 | |||
| 28e59257e1 | |||
| e834e31bbf | |||
| 9dd3fa5f16 | |||
| 257802ea49 | |||
| 0376854c2c | |||
| f24efb78ea | |||
| f27b233a5e | |||
| 969818c735 |
@@ -9,6 +9,7 @@ import './users.js';
|
||||
import './roles.js';
|
||||
import './members.js';
|
||||
import './destinations.js';
|
||||
import './policies.js';
|
||||
import './user-dms.js';
|
||||
import './dropped-messages.js';
|
||||
import './approvals.js';
|
||||
|
||||
@@ -0,0 +1,56 @@
|
||||
import { getAgentGroup } from '../../db/agent-groups.js';
|
||||
import { removeMessagePolicy, setMessagePolicy } from '../../modules/agent-to-agent/db/agent-message-policies.js';
|
||||
import { registerResource } from '../crud.js';
|
||||
|
||||
registerResource({
|
||||
name: 'policy',
|
||||
plural: 'policies',
|
||||
table: 'agent_message_policies',
|
||||
description:
|
||||
'Agent-to-agent approval policy. A row requires every message from one agent to another to be approved by a human before delivery — without un-wiring the connection. No row = free flow. Directed and per-pair: gate both directions with two policies. Operator-only (agents cannot manage their own gates).',
|
||||
idColumn: 'from_agent_group_id',
|
||||
columns: [
|
||||
{ name: 'from_agent_group_id', type: 'string', description: 'Source agent group. References agent_groups.id.' },
|
||||
{ name: 'to_agent_group_id', type: 'string', description: 'Target agent group. References agent_groups.id.' },
|
||||
{
|
||||
name: 'approver',
|
||||
type: 'string',
|
||||
description: 'User-id who approves each gated message (required). Only this user (or an owner) can approve.',
|
||||
},
|
||||
{ name: 'created_at', type: 'string', description: 'Auto-set.' },
|
||||
],
|
||||
operations: { list: 'open' },
|
||||
customOperations: {
|
||||
set: {
|
||||
access: 'approval',
|
||||
description:
|
||||
'Require approval for messages from one agent to another. Use --from <agent-group-id> --to <agent-group-id> --approver <user-id>. Only the named approver (or an owner) can approve.',
|
||||
handler: async (args) => {
|
||||
const from = args.from as string;
|
||||
const to = args.to as string;
|
||||
const approver = args.approver as string;
|
||||
if (!from) throw new Error('--from is required');
|
||||
if (!to) throw new Error('--to is required');
|
||||
if (!approver) throw new Error('--approver is required');
|
||||
if (from === to) throw new Error('--from and --to must differ (self-messages are never gated)');
|
||||
if (!getAgentGroup(from)) throw new Error(`source agent group not found: ${from}`);
|
||||
if (!getAgentGroup(to)) throw new Error(`target agent group not found: ${to}`);
|
||||
|
||||
setMessagePolicy(from, to, approver, new Date().toISOString());
|
||||
return { from_agent_group_id: from, to_agent_group_id: to, approver };
|
||||
},
|
||||
},
|
||||
remove: {
|
||||
access: 'approval',
|
||||
description: 'Remove an approval policy (back to free flow). Use --from <agent-group-id> --to <agent-group-id>.',
|
||||
handler: async (args) => {
|
||||
const from = args.from as string;
|
||||
const to = args.to as string;
|
||||
if (!from) throw new Error('--from is required');
|
||||
if (!to) throw new Error('--to is required');
|
||||
if (!removeMessagePolicy(from, to)) throw new Error('policy not found');
|
||||
return { removed: { from_agent_group_id: from, to_agent_group_id: to } };
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
@@ -4,6 +4,7 @@ import { log } from '../../log.js';
|
||||
import { migration001 } from './001-initial.js';
|
||||
import { migration002 } from './002-chat-sdk-state.js';
|
||||
import { moduleAgentToAgentDestinations } from './module-agent-to-agent-destinations.js';
|
||||
import { moduleAgentMessagePolicies } from './module-agent-message-policies.js';
|
||||
import { migration008 } from './008-dropped-messages.js';
|
||||
import { migration009 } from './009-drop-pending-credentials.js';
|
||||
import { migration010 } from './010-engage-modes.js';
|
||||
@@ -15,6 +16,7 @@ import { migration015 } from './015-cli-scope.js';
|
||||
import { migration016 } from './016-messaging-group-instance.js';
|
||||
import { moduleApprovalsPendingApprovals } from './module-approvals-pending-approvals.js';
|
||||
import { moduleApprovalsTitleOptions } from './module-approvals-title-options.js';
|
||||
import { moduleApprovalsApprover } from './module-approvals-approver.js';
|
||||
|
||||
export interface Migration {
|
||||
version: number;
|
||||
@@ -36,7 +38,9 @@ export const migrations: Migration[] = [
|
||||
migration002,
|
||||
moduleApprovalsPendingApprovals,
|
||||
moduleAgentToAgentDestinations,
|
||||
moduleAgentMessagePolicies,
|
||||
moduleApprovalsTitleOptions,
|
||||
moduleApprovalsApprover,
|
||||
migration008,
|
||||
migration009,
|
||||
migration010,
|
||||
|
||||
@@ -0,0 +1,20 @@
|
||||
import type Database from 'better-sqlite3';
|
||||
|
||||
import type { Migration } from './index.js';
|
||||
|
||||
/** Per-message approval gate on an agent-to-agent connection; no row = free flow. */
|
||||
export const moduleAgentMessagePolicies: Migration = {
|
||||
version: 17,
|
||||
name: 'agent-message-policies',
|
||||
up(db: Database.Database) {
|
||||
db.exec(`
|
||||
CREATE TABLE agent_message_policies (
|
||||
from_agent_group_id TEXT NOT NULL REFERENCES agent_groups(id),
|
||||
to_agent_group_id TEXT NOT NULL REFERENCES agent_groups(id),
|
||||
approver TEXT NOT NULL,
|
||||
created_at TEXT NOT NULL,
|
||||
PRIMARY KEY (from_agent_group_id, to_agent_group_id)
|
||||
);
|
||||
`);
|
||||
},
|
||||
};
|
||||
@@ -0,0 +1,14 @@
|
||||
import type { Migration } from './index.js';
|
||||
|
||||
/**
|
||||
* `approver_user_id` on `pending_approvals`: when an approval names a specific
|
||||
* approver (an a2a message-gate policy's approver), only that exact user may
|
||||
* resolve it. NULL keeps the existing group/owner authorization path.
|
||||
*/
|
||||
export const moduleApprovalsApprover: Migration = {
|
||||
version: 18,
|
||||
name: 'approvals-approver-user-id',
|
||||
up(db) {
|
||||
db.exec(`ALTER TABLE pending_approvals ADD COLUMN approver_user_id TEXT;`);
|
||||
},
|
||||
};
|
||||
+3
-2
@@ -155,11 +155,11 @@ export function createPendingApproval(
|
||||
`INSERT OR IGNORE INTO pending_approvals
|
||||
(approval_id, session_id, request_id, action, payload, created_at,
|
||||
agent_group_id, channel_type, platform_id, platform_message_id, expires_at, status,
|
||||
title, options_json)
|
||||
title, options_json, approver_user_id)
|
||||
VALUES
|
||||
(@approval_id, @session_id, @request_id, @action, @payload, @created_at,
|
||||
@agent_group_id, @channel_type, @platform_id, @platform_message_id, @expires_at, @status,
|
||||
@title, @options_json)`,
|
||||
@title, @options_json, @approver_user_id)`,
|
||||
)
|
||||
.run({
|
||||
session_id: null,
|
||||
@@ -169,6 +169,7 @@ export function createPendingApproval(
|
||||
platform_message_id: null,
|
||||
expires_at: null,
|
||||
status: 'pending',
|
||||
approver_user_id: null,
|
||||
...pa,
|
||||
});
|
||||
return result.changes > 0;
|
||||
|
||||
@@ -29,7 +29,9 @@ import { wakeContainer } from '../../container-runner.js';
|
||||
import { log } from '../../log.js';
|
||||
import { openInboundDb, resolveSession, sessionDir, writeSessionMessage } from '../../session-manager.js';
|
||||
import type { Session } from '../../types.js';
|
||||
import { requestApproval } from '../approvals/index.js';
|
||||
import { hasDestination } from './db/agent-destinations.js';
|
||||
import { getMessagePolicy } from './db/agent-message-policies.js';
|
||||
|
||||
export { isSafeAttachmentName };
|
||||
|
||||
@@ -208,21 +210,87 @@ function resolveTargetSession(msg: RoutableAgentMessage, sourceSession: Session,
|
||||
}
|
||||
|
||||
export async function routeAgentMessage(msg: RoutableAgentMessage, session: Session): Promise<void> {
|
||||
const sourceAgentGroupId = session.agent_group_id;
|
||||
const targetAgentGroupId = msg.platform_id;
|
||||
if (!targetAgentGroupId) {
|
||||
throw new Error(`agent-to-agent message ${msg.id} is missing a target agent group id`);
|
||||
}
|
||||
if (
|
||||
targetAgentGroupId !== session.agent_group_id &&
|
||||
!hasDestination(session.agent_group_id, 'agent', targetAgentGroupId)
|
||||
) {
|
||||
throw new Error(
|
||||
`unauthorized agent-to-agent: ${session.agent_group_id} has no destination for ${targetAgentGroupId}`,
|
||||
);
|
||||
const isSelf = targetAgentGroupId === sourceAgentGroupId;
|
||||
if (!isSelf && !hasDestination(sourceAgentGroupId, 'agent', targetAgentGroupId)) {
|
||||
throw new Error(`unauthorized agent-to-agent: ${sourceAgentGroupId} has no destination for ${targetAgentGroupId}`);
|
||||
}
|
||||
if (!getAgentGroup(targetAgentGroupId)) {
|
||||
throw new Error(`target agent group ${targetAgentGroupId} not found for message ${msg.id}`);
|
||||
}
|
||||
|
||||
// Gated edge: hold the message and return (not throw) so the delivery loop
|
||||
// consumes the outbound row; `applyA2aMessageGate` re-routes it on approve.
|
||||
if (!isSelf) {
|
||||
const policy = getMessagePolicy(sourceAgentGroupId, targetAgentGroupId);
|
||||
if (policy) {
|
||||
const { approver } = policy;
|
||||
const sourceName = getAgentGroup(sourceAgentGroupId)?.name ?? sourceAgentGroupId;
|
||||
const targetName = getAgentGroup(targetAgentGroupId)?.name ?? targetAgentGroupId;
|
||||
await requestApproval({
|
||||
session,
|
||||
agentName: sourceName,
|
||||
action: A2A_MESSAGE_GATE_ACTION,
|
||||
approverUserId: approver,
|
||||
title: 'Message approval',
|
||||
question: buildGateQuestion(sourceName, targetName, msg.content),
|
||||
payload: {
|
||||
id: msg.id,
|
||||
platform_id: targetAgentGroupId,
|
||||
content: msg.content,
|
||||
in_reply_to: msg.in_reply_to,
|
||||
},
|
||||
});
|
||||
log.info('Agent message held for approval', {
|
||||
from: sourceAgentGroupId,
|
||||
to: targetAgentGroupId,
|
||||
msgId: msg.id,
|
||||
});
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
await performAgentRoute(msg, session, targetAgentGroupId);
|
||||
}
|
||||
|
||||
export const A2A_MESSAGE_GATE_ACTION = 'a2a_message_gate';
|
||||
|
||||
const GATE_CARD_BODY_MAX = 1500;
|
||||
|
||||
function parseMessageContent(contentStr: string): { text: string; files: string[] } {
|
||||
try {
|
||||
const parsed = JSON.parse(contentStr) as { text?: unknown; files?: unknown };
|
||||
return {
|
||||
text: typeof parsed.text === 'string' ? parsed.text : '',
|
||||
files: Array.isArray(parsed.files) ? parsed.files.filter((f): f is string => typeof f === 'string') : [],
|
||||
};
|
||||
} catch {
|
||||
return { text: contentStr, files: [] };
|
||||
}
|
||||
}
|
||||
|
||||
function buildGateQuestion(sourceName: string, targetName: string, contentStr: string): string {
|
||||
const { text, files } = parseMessageContent(contentStr);
|
||||
const body = text.length > GATE_CARD_BODY_MAX ? `${text.slice(0, GATE_CARD_BODY_MAX)}… (truncated)` : text;
|
||||
const lines = [`Agent "${sourceName}" wants to send a message to "${targetName}":`, '', body];
|
||||
if (files.length > 0) lines.push('', `Attachments: ${files.join(', ')}`);
|
||||
lines.push('', 'Approve delivery?');
|
||||
return lines.join('\n');
|
||||
}
|
||||
|
||||
/**
|
||||
* Cross-session route: pick the target session, forward files, write to its
|
||||
* inbound DB, wake it. Authorization is the caller's responsibility.
|
||||
*/
|
||||
export async function performAgentRoute(
|
||||
msg: RoutableAgentMessage,
|
||||
session: Session,
|
||||
targetAgentGroupId: string,
|
||||
): Promise<void> {
|
||||
const targetSession = resolveTargetSession(msg, session, targetAgentGroupId);
|
||||
const a2aMsgId = `a2a-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
|
||||
|
||||
|
||||
@@ -36,6 +36,7 @@
|
||||
*/
|
||||
import type { AgentDestination } from '../../../types.js';
|
||||
import { getDb } from '../../../db/connection.js';
|
||||
import { deletePoliciesTouching, removeMessagePolicy } from './agent-message-policies.js';
|
||||
|
||||
/**
|
||||
* ⚠️ Caller responsibility: after this returns, call
|
||||
@@ -89,9 +90,16 @@ export function hasDestination(agentGroupId: string, targetType: 'channel' | 'ag
|
||||
* so the deletion propagates to the running container's inbound.db.
|
||||
*/
|
||||
export function deleteDestination(agentGroupId: string, localName: string): void {
|
||||
// Resolve the target first so we can drop a matching policy for this edge (no ghost gate on re-wire).
|
||||
const row = getDb()
|
||||
.prepare('SELECT target_type, target_id FROM agent_destinations WHERE agent_group_id = ? AND local_name = ?')
|
||||
.get(agentGroupId, localName) as { target_type: string; target_id: string } | undefined;
|
||||
getDb()
|
||||
.prepare('DELETE FROM agent_destinations WHERE agent_group_id = ? AND local_name = ?')
|
||||
.run(agentGroupId, localName);
|
||||
if (row?.target_type === 'agent') {
|
||||
removeMessagePolicy(agentGroupId, row.target_id);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -108,6 +116,7 @@ export function deleteAllDestinationsTouching(agentGroupId: string): void {
|
||||
getDb()
|
||||
.prepare('DELETE FROM agent_destinations WHERE agent_group_id = ? OR (target_type = ? AND target_id = ?)')
|
||||
.run(agentGroupId, 'agent', agentGroupId);
|
||||
deletePoliciesTouching(agentGroupId);
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
/** Per-message approval policies for agent-to-agent connections; no row = free flow. */
|
||||
import type { AgentMessagePolicy } from '../../../types.js';
|
||||
import { getDb } from '../../../db/connection.js';
|
||||
|
||||
export function getMessagePolicy(fromAgentGroupId: string, toAgentGroupId: string): AgentMessagePolicy | undefined {
|
||||
return getDb()
|
||||
.prepare('SELECT * FROM agent_message_policies WHERE from_agent_group_id = ? AND to_agent_group_id = ?')
|
||||
.get(fromAgentGroupId, toAgentGroupId) as AgentMessagePolicy | undefined;
|
||||
}
|
||||
|
||||
export function setMessagePolicy(
|
||||
fromAgentGroupId: string,
|
||||
toAgentGroupId: string,
|
||||
approver: string,
|
||||
createdAt: string,
|
||||
): void {
|
||||
getDb()
|
||||
.prepare(
|
||||
`INSERT INTO agent_message_policies (from_agent_group_id, to_agent_group_id, approver, created_at)
|
||||
VALUES (@from_agent_group_id, @to_agent_group_id, @approver, @created_at)
|
||||
ON CONFLICT (from_agent_group_id, to_agent_group_id) DO UPDATE SET approver = excluded.approver`,
|
||||
)
|
||||
.run({ from_agent_group_id: fromAgentGroupId, to_agent_group_id: toAgentGroupId, approver, created_at: createdAt });
|
||||
}
|
||||
|
||||
export function removeMessagePolicy(fromAgentGroupId: string, toAgentGroupId: string): boolean {
|
||||
const info = getDb()
|
||||
.prepare('DELETE FROM agent_message_policies WHERE from_agent_group_id = ? AND to_agent_group_id = ?')
|
||||
.run(fromAgentGroupId, toAgentGroupId);
|
||||
return info.changes > 0;
|
||||
}
|
||||
|
||||
/** Delete every policy touching this agent group, so none outlives its connection. */
|
||||
export function deletePoliciesTouching(agentGroupId: string): void {
|
||||
getDb()
|
||||
.prepare('DELETE FROM agent_message_policies WHERE from_agent_group_id = ? OR to_agent_group_id = ?')
|
||||
.run(agentGroupId, agentGroupId);
|
||||
}
|
||||
@@ -22,7 +22,11 @@
|
||||
*/
|
||||
import { registerDeliveryAction } from '../../delivery.js';
|
||||
import { registerApprovalHandler } from '../approvals/index.js';
|
||||
import { A2A_MESSAGE_GATE_ACTION } from './agent-route.js';
|
||||
import { applyCreateAgent, handleCreateAgent } from './create-agent.js';
|
||||
import { applyA2aMessageGate } from './message-gate.js';
|
||||
|
||||
registerDeliveryAction('create_agent', handleCreateAgent);
|
||||
registerApprovalHandler('create_agent', applyCreateAgent);
|
||||
|
||||
registerApprovalHandler(A2A_MESSAGE_GATE_ACTION, applyA2aMessageGate);
|
||||
|
||||
@@ -0,0 +1,193 @@
|
||||
import Database from 'better-sqlite3';
|
||||
import fs from 'fs';
|
||||
import { describe, expect, it, beforeEach, afterEach, vi } from 'vitest';
|
||||
|
||||
import { routeAgentMessage } from './agent-route.js';
|
||||
import { createDestination, deleteDestination, deleteAllDestinationsTouching } from './db/agent-destinations.js';
|
||||
import { getMessagePolicy, removeMessagePolicy, setMessagePolicy } from './db/agent-message-policies.js';
|
||||
import { applyA2aMessageGate } from './message-gate.js';
|
||||
import { initTestDb, closeDb, runMigrations, createAgentGroup } from '../../db/index.js';
|
||||
import { getDb } from '../../db/connection.js';
|
||||
import { createSession } from '../../db/sessions.js';
|
||||
import { requestApproval } from '../approvals/index.js';
|
||||
import { initSessionFolder, inboundDbPath } from '../../session-manager.js';
|
||||
import type { Session } from '../../types.js';
|
||||
|
||||
vi.mock('../../container-runner.js', () => ({
|
||||
wakeContainer: vi.fn().mockResolvedValue(undefined),
|
||||
isContainerRunning: vi.fn().mockReturnValue(false),
|
||||
getActiveContainerCount: vi.fn().mockReturnValue(0),
|
||||
killContainer: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock('../approvals/index.js', async (importActual) => {
|
||||
const actual = await importActual<typeof import('../approvals/index.js')>();
|
||||
return { ...actual, requestApproval: vi.fn().mockResolvedValue(undefined) };
|
||||
});
|
||||
|
||||
vi.mock('../../config.js', async () => {
|
||||
const actual = await vi.importActual('../../config.js');
|
||||
return { ...actual, DATA_DIR: '/tmp/nanoclaw-test-a2a-gate' };
|
||||
});
|
||||
|
||||
const TEST_DIR = '/tmp/nanoclaw-test-a2a-gate';
|
||||
const A = 'ag-A';
|
||||
const B = 'ag-B';
|
||||
|
||||
function now(): string {
|
||||
return new Date().toISOString();
|
||||
}
|
||||
|
||||
function policyCount(): number {
|
||||
return (getDb().prepare('SELECT COUNT(*) AS n FROM agent_message_policies').get() as { n: number }).n;
|
||||
}
|
||||
|
||||
function readInbound(agentGroupId: string, sessionId: string) {
|
||||
const db = new Database(inboundDbPath(agentGroupId, sessionId), { readonly: true });
|
||||
const rows = db.prepare('SELECT id, platform_id, content FROM messages_in ORDER BY seq').all() as Array<{
|
||||
id: string;
|
||||
platform_id: string | null;
|
||||
content: string;
|
||||
}>;
|
||||
db.close();
|
||||
return rows;
|
||||
}
|
||||
|
||||
function makeSession(id: string, agentGroupId: string): Session {
|
||||
return {
|
||||
id,
|
||||
agent_group_id: agentGroupId,
|
||||
messaging_group_id: null,
|
||||
thread_id: null,
|
||||
agent_provider: null,
|
||||
status: 'active',
|
||||
container_status: 'stopped',
|
||||
last_active: null,
|
||||
created_at: now(),
|
||||
};
|
||||
}
|
||||
|
||||
describe('agent message policies', () => {
|
||||
let SA: Session;
|
||||
let SB: Session;
|
||||
|
||||
beforeEach(() => {
|
||||
if (fs.existsSync(TEST_DIR)) fs.rmSync(TEST_DIR, { recursive: true });
|
||||
fs.mkdirSync(TEST_DIR, { recursive: true });
|
||||
const db = initTestDb();
|
||||
runMigrations(db);
|
||||
vi.mocked(requestApproval).mockClear();
|
||||
|
||||
createAgentGroup({ id: A, name: 'A', folder: 'a', agent_provider: null, created_at: now() });
|
||||
createAgentGroup({ id: B, name: 'B', folder: 'b', agent_provider: null, created_at: now() });
|
||||
SA = makeSession('sess-A', A);
|
||||
SB = makeSession('sess-B', B);
|
||||
createSession(SA);
|
||||
createSession(SB);
|
||||
initSessionFolder(A, SA.id);
|
||||
initSessionFolder(B, SB.id);
|
||||
// A→B connection wired.
|
||||
createDestination({ agent_group_id: A, local_name: 'b', target_type: 'agent', target_id: B, created_at: now() });
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
closeDb();
|
||||
if (fs.existsSync(TEST_DIR)) fs.rmSync(TEST_DIR, { recursive: true });
|
||||
});
|
||||
|
||||
// ── policy table round-trip ──
|
||||
|
||||
it('set / get / remove round-trip, incl. approver', () => {
|
||||
expect(getMessagePolicy(A, B)).toBeUndefined();
|
||||
|
||||
setMessagePolicy(A, B, 'telegram:sam', now());
|
||||
expect(getMessagePolicy(A, B)).toMatchObject({
|
||||
from_agent_group_id: A,
|
||||
to_agent_group_id: B,
|
||||
approver: 'telegram:sam',
|
||||
});
|
||||
expect(policyCount()).toBe(1);
|
||||
|
||||
// Upsert updates the approver without inserting a duplicate row.
|
||||
setMessagePolicy(A, B, 'telegram:dana', now());
|
||||
expect(getMessagePolicy(A, B)!.approver).toBe('telegram:dana');
|
||||
expect(policyCount()).toBe(1);
|
||||
|
||||
expect(removeMessagePolicy(A, B)).toBe(true);
|
||||
expect(getMessagePolicy(A, B)).toBeUndefined();
|
||||
expect(removeMessagePolicy(A, B)).toBe(false);
|
||||
});
|
||||
|
||||
// ── gate behavior in routeAgentMessage ──
|
||||
|
||||
it('no policy → routes normally, no approval requested', async () => {
|
||||
await routeAgentMessage(
|
||||
{ id: 'm1', platform_id: B, content: JSON.stringify({ text: 'hi B' }), in_reply_to: null },
|
||||
SA,
|
||||
);
|
||||
expect(readInbound(B, SB.id)).toHaveLength(1);
|
||||
expect(requestApproval).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('policy present → holds the message and requests approval from the policy approver scoped to the target', async () => {
|
||||
setMessagePolicy(A, B, 'telegram:dana', now());
|
||||
|
||||
await routeAgentMessage(
|
||||
{ id: 'm2', platform_id: B, content: JSON.stringify({ text: 'sensitive' }), in_reply_to: null },
|
||||
SA,
|
||||
);
|
||||
|
||||
// Held: nothing routed to B.
|
||||
expect(readInbound(B, SB.id)).toHaveLength(0);
|
||||
// One approval requested, to the policy's approver, scoped to the target group.
|
||||
expect(requestApproval).toHaveBeenCalledTimes(1);
|
||||
const opts = vi.mocked(requestApproval).mock.calls[0][0];
|
||||
expect(opts.action).toBe('a2a_message_gate');
|
||||
expect(opts.approverUserId).toBe('telegram:dana');
|
||||
expect(opts.payload).toMatchObject({ id: 'm2', platform_id: B });
|
||||
expect(JSON.parse(String(opts.payload.content)).text).toBe('sensitive');
|
||||
});
|
||||
|
||||
it('self-message is never gated even if a policy row somehow exists', async () => {
|
||||
setMessagePolicy(A, A, 'telegram:dana', now()); // pathological, but must be ignored
|
||||
await routeAgentMessage(
|
||||
{ id: 'self', platform_id: A, content: JSON.stringify({ text: 'note' }), in_reply_to: null },
|
||||
SA,
|
||||
);
|
||||
expect(requestApproval).not.toHaveBeenCalled();
|
||||
expect(readInbound(A, SA.id)).toHaveLength(1);
|
||||
});
|
||||
|
||||
// ── approve handler re-routes the held message ──
|
||||
|
||||
it('applyA2aMessageGate delivers the held message to the target', async () => {
|
||||
const notify = vi.fn();
|
||||
await applyA2aMessageGate({
|
||||
session: SA,
|
||||
userId: 'slack:dana',
|
||||
notify,
|
||||
payload: { id: 'held-1', platform_id: B, content: JSON.stringify({ text: 'approved!' }), in_reply_to: null },
|
||||
});
|
||||
|
||||
const bRows = readInbound(B, SB.id);
|
||||
expect(bRows).toHaveLength(1);
|
||||
expect(JSON.parse(bRows[0].content).text).toBe('approved!');
|
||||
expect(notify).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
// ── ghost-gate cleanup ──
|
||||
|
||||
it('deleting the connection drops its policy', () => {
|
||||
setMessagePolicy(A, B, 'telegram:dana', now());
|
||||
deleteDestination(A, 'b'); // removes the A→B agent destination
|
||||
expect(getMessagePolicy(A, B)).toBeUndefined();
|
||||
});
|
||||
|
||||
it('deleteAllDestinationsTouching drops policies on both sides', () => {
|
||||
setMessagePolicy(A, B, 'telegram:dana', now());
|
||||
setMessagePolicy(B, A, 'telegram:dana', now());
|
||||
deleteAllDestinationsTouching(A);
|
||||
expect(getMessagePolicy(A, B)).toBeUndefined();
|
||||
expect(getMessagePolicy(B, A)).toBeUndefined();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,27 @@
|
||||
/** Approve handler for a held a2a message. (Reject is handled by the generic response-handler path.) */
|
||||
import { log } from '../../log.js';
|
||||
import type { ApprovalHandler } from '../approvals/index.js';
|
||||
import { performAgentRoute, type RoutableAgentMessage } from './agent-route.js';
|
||||
|
||||
export const applyA2aMessageGate: ApprovalHandler = async ({ session, payload, notify }) => {
|
||||
const { id, platform_id, content, in_reply_to } = payload;
|
||||
if (typeof platform_id !== 'string' || !platform_id) {
|
||||
notify('Message approved but the target agent group was missing from the request.');
|
||||
log.warn('a2a_message_gate apply: missing target', { sessionId: session.id });
|
||||
return;
|
||||
}
|
||||
|
||||
const msg: RoutableAgentMessage = {
|
||||
id: typeof id === 'string' ? id : `a2a-gate-${Date.now()}`,
|
||||
platform_id,
|
||||
content: typeof content === 'string' ? content : '',
|
||||
in_reply_to: typeof in_reply_to === 'string' ? in_reply_to : null,
|
||||
};
|
||||
|
||||
await performAgentRoute(msg, session, platform_id);
|
||||
log.info('Held agent message delivered after approval', {
|
||||
from: session.agent_group_id,
|
||||
to: platform_id,
|
||||
msgId: msg.id,
|
||||
});
|
||||
};
|
||||
@@ -197,6 +197,8 @@ export interface RequestApprovalOptions {
|
||||
title: string;
|
||||
/** Card body shown to the admin. */
|
||||
question: string;
|
||||
/** Deliver the card to this specific user instead of all of the session group's admins. */
|
||||
approverUserId?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -206,9 +208,9 @@ export interface RequestApprovalOptions {
|
||||
* approval handler for this action via the response dispatcher.
|
||||
*/
|
||||
export async function requestApproval(opts: RequestApprovalOptions): Promise<void> {
|
||||
const { session, action, payload, title, question, agentName } = opts;
|
||||
const { session, action, payload, title, question, agentName, approverUserId } = opts;
|
||||
|
||||
const approvers = pickApprover(session.agent_group_id);
|
||||
const approvers = approverUserId ? [approverUserId] : pickApprover(session.agent_group_id);
|
||||
if (approvers.length === 0) {
|
||||
notifyAgent(session, `${action} failed: no owner or admin configured to approve.`);
|
||||
return;
|
||||
@@ -235,6 +237,7 @@ export async function requestApproval(opts: RequestApprovalOptions): Promise<voi
|
||||
created_at: new Date().toISOString(),
|
||||
title,
|
||||
options_json: JSON.stringify(normalizedOptions),
|
||||
approver_user_id: approverUserId ?? null,
|
||||
});
|
||||
|
||||
const adapter = getDeliveryAdapter();
|
||||
|
||||
@@ -161,4 +161,47 @@ describe('approval response authorization', () => {
|
||||
expect(handler).toHaveBeenCalledTimes(1);
|
||||
expect(getPendingApproval('appr-3')).toBeUndefined();
|
||||
});
|
||||
|
||||
it('an approval with approver_user_id is resolvable by that user, not a non-assignee', async () => {
|
||||
const { registerApprovalHandler } = await import('./primitive.js');
|
||||
const { handleApprovalsResponse } = await import('./response-handler.js');
|
||||
const handler = vi.fn().mockResolvedValue(undefined);
|
||||
registerApprovalHandler('assigned_approver_action', handler);
|
||||
|
||||
createPendingApproval({
|
||||
approval_id: 'appr-4',
|
||||
session_id: 'sess-1',
|
||||
request_id: 'appr-4',
|
||||
action: 'assigned_approver_action',
|
||||
payload: JSON.stringify({}),
|
||||
created_at: now(),
|
||||
title: 'Assigned approval',
|
||||
options_json: JSON.stringify([]),
|
||||
approver_user_id: 'telegram:dana',
|
||||
});
|
||||
|
||||
// A non-assignee (no global/owner role) cannot resolve it.
|
||||
await handleApprovalsResponse({
|
||||
questionId: 'appr-4',
|
||||
value: 'approve',
|
||||
userId: 'stranger',
|
||||
channelType: 'telegram',
|
||||
platformId: 'dm-stranger',
|
||||
threadId: null,
|
||||
});
|
||||
expect(handler).not.toHaveBeenCalled();
|
||||
expect(getPendingApproval('appr-4')).toBeDefined();
|
||||
|
||||
// The named approver resolves it.
|
||||
await handleApprovalsResponse({
|
||||
questionId: 'appr-4',
|
||||
value: 'approve',
|
||||
userId: 'dana',
|
||||
channelType: 'telegram',
|
||||
platformId: 'dm-dana',
|
||||
threadId: null,
|
||||
});
|
||||
expect(handler).toHaveBeenCalledTimes(1);
|
||||
expect(getPendingApproval('appr-4')).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -125,6 +125,11 @@ function isAuthorizedApprovalClick(approval: PendingApproval, payload: ResponseP
|
||||
const userId = namespacedUserId(payload);
|
||||
if (!userId) return false;
|
||||
|
||||
// An approval may name a specific approver; only that exact user may resolve it.
|
||||
if (approval.approver_user_id) {
|
||||
return userId === approval.approver_user_id;
|
||||
}
|
||||
|
||||
const agentGroupId =
|
||||
approval.agent_group_id ?? (approval.session_id ? getSession(approval.session_id)?.agent_group_id : null);
|
||||
|
||||
|
||||
@@ -204,6 +204,8 @@ export interface PendingApproval {
|
||||
status: 'pending' | 'approved' | 'rejected' | 'expired';
|
||||
title: string;
|
||||
options_json: string;
|
||||
/** When set, only this exact user may resolve the approval. */
|
||||
approver_user_id: string | null;
|
||||
}
|
||||
|
||||
// ── Agent destinations (central DB) ──
|
||||
@@ -215,3 +217,10 @@ export interface AgentDestination {
|
||||
target_id: string;
|
||||
created_at: string;
|
||||
}
|
||||
|
||||
export interface AgentMessagePolicy {
|
||||
from_agent_group_id: string;
|
||||
to_agent_group_id: string;
|
||||
approver: string;
|
||||
created_at: string;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user