fix: report DSH runtime events to AgentGuard Cloud

package.json

@@ -22,6 +22,8 @@
     "agentguard-mcp": "./dist/mcp-server.js"
   },
   "scripts": {
+    "clean": "node -e \"require('node:fs').rmSync('dist', { recursive: true, force: true })\"",
+    "prebuild": "npm run clean",
     "build": "tsc",
     "build:mcpb": "bash scripts/build-mcpb.sh",
     "start": "node dist/mcp-server.js",

src/dsh/runtime.ts

@@ -1,7 +1,7 @@
 import { AgentGuardCloudClient } from '../cloud/client.js';
 import { loadConfig, type AgentGuardConfig } from '../config.js';
 import { isAbsolute, resolve } from 'node:path';
-import { writeAuditLog } from '../runtime/audit.js';
+import { flushEventSpool, spoolEvent, writeAuditLog } from '../runtime/audit.js';
 import {
   evaluateRuntimeAction,
   type RuntimeEvaluation,
@@ -142,6 +142,7 @@ const AGENTGUARD_DSH_TOOLS = new Set([
 ]);
 const DSH_OWNER_ID_PATTERN = /^[A-Za-z0-9@][A-Za-z0-9@._/:-]{0,159}$/;
 const MAX_DSH_TOOL_OWNER_BINDINGS = 500;
+const dshSpoolLocks = new Map<string, Promise<void>>();
 
 /** Validate and snapshot operator-authored DSH tool ownership bindings. */
 export function normalizeDshRuntimeAttribution(value: unknown): DshRuntimeAttributionConfig {
@@ -305,13 +306,19 @@ async function evaluateAndAuditDshAction(
   runtimeMode: Exclude<DshRuntimeMode, 'off'>,
   enforcementApplied: boolean | 'block-only'
 ): Promise<DshRuntimeObservation> {
+  const client = new AgentGuardCloudClient(config);
+  if (client.connected) {
+    await withDshSpoolLock(config.eventSpoolPath, () =>
+      flushEventSpool(config.eventSpoolPath, events => client.ingestEvents(events))
+    ).catch(() => undefined);
+  }
   const evaluate = dependencies.evaluate ?? evaluateRuntimeAction;
   const sharedEvaluation = await evaluate({
     action,
     policyCachePath: config.policyCachePath,
     fetchPolicy: dependencies.fetchPolicyFor
       ? dependencies.fetchPolicyFor(config)
-      : defaultFetchPolicy(config),
+      : defaultFetchPolicy(client),
   });
   const guardedEvaluation = applyUnknownToolDecision(
     sharedEvaluation,
@@ -352,6 +359,7 @@ async function evaluateAndAuditDshAction(
   } catch {
     // Phase 2A is fail-open: audit I/O cannot change DSH tool behavior.
   }
+  if (client.connected) await reportDshEvent(client, config, event);
   return { action, evaluation, event };
 }
 
@@ -475,11 +483,47 @@ export function createDshPostExecuteProtector(
   };
 }
 
-function defaultFetchPolicy(config: AgentGuardConfig): (() => Promise<import('../runtime/types.js').EffectiveRuntimePolicy | null>) | undefined {
-  const client = new AgentGuardCloudClient(config);
+function defaultFetchPolicy(client: AgentGuardCloudClient): (() => Promise<import('../runtime/types.js').EffectiveRuntimePolicy | null>) | undefined {
   return client.connected ? () => client.fetchEffectivePolicy() : undefined;
 }
 
+async function reportDshEvent(
+  client: AgentGuardCloudClient,
+  config: AgentGuardConfig,
+  event: RuntimeAuditEvent
+): Promise<void> {
+  try {
+    await client.ingestEvents([event]);
+  } catch {
+    await withDshSpoolLock(config.eventSpoolPath, () => {
+      try {
+        spoolEvent(config.eventSpoolPath, event);
+      } catch {
+        // Cloud reporting is best-effort and cannot change DSH tool behavior.
+      }
+    });
+  }
+}
+
+async function withDshSpoolLock<T>(
+  spoolPath: string,
+  operation: () => T | Promise<T>
+): Promise<T> {
+  const previous = dshSpoolLocks.get(spoolPath) ?? Promise.resolve();
+  let release: () => void = () => undefined;
+  const current = new Promise<void>(resolvePromise => {
+    release = resolvePromise;
+  });
+  dshSpoolLocks.set(spoolPath, current);
+  await previous.catch(() => undefined);
+  try {
+    return await operation();
+  } finally {
+    release();
+    if (dshSpoolLocks.get(spoolPath) === current) dshSpoolLocks.delete(spoolPath);
+  }
+}
+
 function actionInput(actionType: RuntimeActionType, args: Record<string, unknown> | null, raw: unknown): string {
   if (args) {
     if (actionType === 'shell') return firstString(args.command, args.cmd, args.script, args.code, args.input) || stableJson(raw);

src/tests/cli-init.test.ts

@@ -356,7 +356,12 @@ describe('init CLI', () => {
 
     const { stdout } = await execFileAsync(process.execPath, [cliPath, 'init', '--agent', 'auto', '--force'], {
       cwd,
-      env: { ...process.env, AGENTGUARD_HOME: home },
+      env: {
+        ...process.env,
+        AGENTGUARD_HOME: home,
+        DSH_HOME: join(home, 'missing-dsh-home'),
+        DSH_SHELL: '0',
+      },
     });
 
     const config = JSON.parse(readFileSync(join(home, 'config.json'), 'utf8')) as {
@@ -530,7 +535,12 @@ describe('init CLI', () => {
 
     const { stdout, stderr } = await execFileAsync(process.execPath, [cliPath, 'init', '--agent', 'auto', '--force'], {
       cwd,
-      env: { ...process.env, AGENTGUARD_HOME: home },
+      env: {
+        ...process.env,
+        AGENTGUARD_HOME: home,
+        DSH_HOME: join(home, 'missing-dsh-home'),
+        DSH_SHELL: '0',
+      },
     });
 
     const config = JSON.parse(readFileSync(join(home, 'config.json'), 'utf8')) as {

src/tests/dsh-runtime.test.ts

@@ -1,5 +1,10 @@
 import { describe, it } from 'node:test';
 import assert from 'node:assert/strict';
+import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
+import { createServer, type Server } from 'node:http';
+import { type AddressInfo } from 'node:net';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
 import type { AgentGuardConfig } from '../config.js';
 import {
   buildDshRuntimeAction,
@@ -56,6 +61,42 @@ function decision(value: RuntimeDecision['decision'] = 'block'): RuntimeDecision
   };
 }
 
+interface CloudRequest {
+  method: string;
+  path: string;
+  body: unknown;
+}
+
+async function startCloudServer(statuses: number[] = []): Promise<{
+  url: string;
+  requests: CloudRequest[];
+  close: () => Promise<void>;
+}> {
+  const requests: CloudRequest[] = [];
+  const server: Server = createServer(async (request, response) => {
+    const chunks: Buffer[] = [];
+    for await (const chunk of request) chunks.push(chunk as Buffer);
+    const rawBody = Buffer.concat(chunks).toString('utf8');
+    requests.push({
+      method: request.method || '',
+      path: request.url || '',
+      body: rawBody ? JSON.parse(rawBody) : undefined,
+    });
+    response.statusCode = statuses.shift() ?? 200;
+    response.setHeader('content-type', 'application/json');
+    response.end(JSON.stringify({ success: true, data: {} }));
+  });
+  await new Promise<void>((resolvePromise) => server.listen(0, '127.0.0.1', resolvePromise));
+  const address = server.address() as AddressInfo;
+  return {
+    url: `http://127.0.0.1:${address.port}`,
+    requests,
+    close: () => new Promise<void>((resolvePromise, reject) => {
+      server.close(error => error ? reject(error) : resolvePromise());
+    }),
+  };
+}
+
 describe('DSH runtime Phase 2A observer', () => {
   it('normalizes common DSH tools into the shared RuntimeAction vocabulary', () => {
     assert.equal(mapDshToolToRuntimeAction('bash'), 'shell');
@@ -198,6 +239,193 @@ describe('DSH runtime Phase 2A observer', () => {
     assert.equal(written[0].path, config.auditPath);
   });
 
+  it('uploads DSH audit events through the connected Cloud client', async () => {
+    const cloud = await startCloudServer();
+    const directory = mkdtempSync(join(tmpdir(), 'agentguard-dsh-cloud-success-'));
+    try {
+      const observed = await observeDshToolCall(execution({
+        arguments: { command: 'curl https://example.com?token=secret-value' },
+      }), {
+        loadAgentGuardConfig: () => ({
+          ...config,
+          cloudUrl: cloud.url,
+          apiKey: 'ag_live_dsh_cloud_test',
+          eventSpoolPath: join(directory, 'events.jsonl'),
+        }),
+        fetchPolicyFor: () => undefined,
+        evaluate: async () => ({ decision: decision('block'), policySource: 'default' }),
+        writeAudit() {},
+      });
+
+      assert.ok(observed);
+      assert.equal(cloud.requests.length, 1);
+      assert.equal(cloud.requests[0].method, 'POST');
+      assert.equal(cloud.requests[0].path, '/api/v1/events/ingest');
+      assert.equal((cloud.requests[0].body as any).events[0].actionId, 'act-test');
+      assert.equal((cloud.requests[0].body as any).events[0].agentHost, 'dsh');
+      assert.doesNotMatch(JSON.stringify(cloud.requests[0].body), /secret-value/);
+    } finally {
+      await cloud.close();
+      rmSync(directory, { recursive: true, force: true });
+    }
+  });
+
+  it('spools the DSH audit event when connected Cloud ingest fails', async () => {
+    const cloud = await startCloudServer([503]);
+    const directory = mkdtempSync(join(tmpdir(), 'agentguard-dsh-cloud-failure-'));
+    const spoolPath = join(directory, 'events.jsonl');
+    try {
+      const observed = await observeDshToolCall(execution(), {
+        loadAgentGuardConfig: () => ({
+          ...config,
+          cloudUrl: cloud.url,
+          apiKey: 'ag_live_dsh_cloud_test',
+          eventSpoolPath: spoolPath,
+        }),
+        fetchPolicyFor: () => undefined,
+        evaluate: async () => ({ decision: decision('block'), policySource: 'default' }),
+        writeAudit() {},
+      });
+
+      assert.ok(observed);
+      assert.equal(cloud.requests.length, 1);
+      const spooled = readFileSync(spoolPath, 'utf8').trim().split('\n').map(line => JSON.parse(line));
+      assert.equal(spooled.length, 1);
+      assert.equal(spooled[0].actionId, 'act-test');
+      assert.equal(spooled[0].agentHost, 'dsh');
+    } finally {
+      await cloud.close();
+      rmSync(directory, { recursive: true, force: true });
+    }
+  });
+
+  it('preserves the DSH result when Cloud ingest and spool writes both fail', async () => {
+    const cloud = await startCloudServer([503]);
+    const directory = mkdtempSync(join(tmpdir(), 'agentguard-dsh-cloud-fail-open-'));
+    try {
+      const protector = createDshPreExecuteProtector({
+        loadAgentGuardConfig: () => ({
+          ...config,
+          cloudUrl: cloud.url,
+          apiKey: 'ag_live_dsh_cloud_test',
+          eventSpoolPath: directory,
+        }),
+        fetchPolicyFor: () => undefined,
+        evaluate: async () => ({ decision: decision('allow'), policySource: 'default' }),
+        writeAudit() {},
+      });
+      const downstream = { kind: 'allow' as const };
+      const protectedDecision = await protector(execution(), async () => downstream);
+
+      assert.deepEqual(protectedDecision, downstream);
+      assert.equal(cloud.requests.length, 1);
+    } finally {
+      await cloud.close();
+      rmSync(directory, { recursive: true, force: true });
+    }
+  });
+
+  it('flushes queued DSH events before uploading the current event', async () => {
+    const cloud = await startCloudServer();
+    const directory = mkdtempSync(join(tmpdir(), 'agentguard-dsh-cloud-retry-'));
+    const spoolPath = join(directory, 'events.jsonl');
+    writeFileSync(spoolPath, `${JSON.stringify({
+      actionId: 'act-queued',
+      sessionId: 'session-queued',
+      agentHost: 'dsh',
+      actionType: 'shell',
+      toolName: 'bash',
+      input: 'echo queued',
+      decision: 'warn',
+      riskScore: 40,
+      riskLevel: 'medium',
+      reasons: [],
+      policyVersion: 'runtime-test',
+    })}\n`);
+    try {
+      const observed = await observeDshToolCall(execution(), {
+        loadAgentGuardConfig: () => ({
+          ...config,
+          cloudUrl: cloud.url,
+          apiKey: 'ag_live_dsh_cloud_test',
+          eventSpoolPath: spoolPath,
+        }),
+        fetchPolicyFor: () => undefined,
+        evaluate: async () => {
+          assert.equal(cloud.requests.length, 1);
+          assert.equal((cloud.requests[0].body as any).events[0].actionId, 'act-queued');
+          return { decision: decision('block'), policySource: 'default' };
+        },
+        writeAudit() {},
+      });
+
+      assert.ok(observed);
+      assert.equal(cloud.requests.length, 2);
+      assert.equal((cloud.requests[0].body as any).events[0].actionId, 'act-queued');
+      assert.equal((cloud.requests[1].body as any).events[0].actionId, 'act-test');
+      assert.equal(existsSync(spoolPath), false);
+    } finally {
+      await cloud.close();
+      rmSync(directory, { recursive: true, force: true });
+    }
+  });
+
+  it('flushes each queued DSH event once across concurrent tool calls', async () => {
+    const cloud = await startCloudServer();
+    const directory = mkdtempSync(join(tmpdir(), 'agentguard-dsh-cloud-concurrent-'));
+    const spoolPath = join(directory, 'events.jsonl');
+    writeFileSync(spoolPath, `${JSON.stringify({
+      actionId: 'act-queued',
+      sessionId: 'session-queued',
+      agentHost: 'dsh',
+      actionType: 'shell',
+      toolName: 'bash',
+      input: 'echo queued',
+      decision: 'warn',
+      riskScore: 40,
+      riskLevel: 'medium',
+      reasons: [],
+      policyVersion: 'runtime-test',
+    })}\n`);
+    const connectedConfig: AgentGuardConfig = {
+      ...config,
+      cloudUrl: cloud.url,
+      apiKey: 'ag_live_dsh_cloud_test',
+      eventSpoolPath: spoolPath,
+    };
+    const evaluate = async ({ action }: any) => ({
+      decision: { ...decision('block'), actionId: `act-${action.metadata.callId}` },
+      policySource: 'default' as const,
+    });
+    try {
+      await Promise.all([
+        observeDshToolCall(execution({ callId: 'current-1' }), {
+          loadAgentGuardConfig: () => connectedConfig,
+          fetchPolicyFor: () => undefined,
+          evaluate,
+          writeAudit() {},
+        }),
+        observeDshToolCall(execution({ callId: 'current-2' }), {
+          loadAgentGuardConfig: () => connectedConfig,
+          fetchPolicyFor: () => undefined,
+          evaluate,
+          writeAudit() {},
+        }),
+      ]);
+
+      const actionIds = cloud.requests.flatMap(request =>
+        (request.body as any).events.map((event: any) => event.actionId)
+      );
+      assert.equal(actionIds.filter(actionId => actionId === 'act-queued').length, 1);
+      assert.equal(actionIds.filter(actionId => actionId === 'act-current-1').length, 1);
+      assert.equal(actionIds.filter(actionId => actionId === 'act-current-2').length, 1);
+      assert.equal(existsSync(spoolPath), false);
+    } finally {
+      await cloud.close();
+      rmSync(directory, { recursive: true, force: true });
+    }
+  });
+
   it('keeps response anomaly semantics stable across the DSH fixture corpus', async () => {
     const responseCodes = new Set([
       'RESPONSE_XSS_ECHO', 'RESPONSE_ERROR_DISCLOSURE', 'RESPONSE_MALICIOUS_SCRIPT',