From 71daf76c7e8dcea1d6794cd7359d564191b87952 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 6 Aug 2026 15:09:28 +0000 Subject: [PATCH 1/2] Initial plan From 6027491d4788fed1fe0739b1035f6c1a9372e1ba Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 6 Aug 2026 15:15:53 +0000 Subject: [PATCH 2/2] test: share ingress conformance harness --- src/bounded-agent/ingress-conformance.test.ts | 157 ++++-------------- src/bounded-query/ingress-conformance.test.ts | 152 ++++------------- .../ingress-conformance-test-harness.ts | 154 +++++++++++++++++ 3 files changed, 221 insertions(+), 242 deletions(-) create mode 100644 src/test-helpers/ingress-conformance-test-harness.ts diff --git a/src/bounded-agent/ingress-conformance.test.ts b/src/bounded-agent/ingress-conformance.test.ts index bc303c6ba..5183467fa 100644 --- a/src/bounded-agent/ingress-conformance.test.ts +++ b/src/bounded-agent/ingress-conformance.test.ts @@ -1,13 +1,18 @@ -import * as fs from 'fs'; -import * as http from 'http'; import * as net from 'net'; -import * as os from 'os'; import * as path from 'path'; -import type { AddressInfo } from 'net'; +import { + CANONICAL_ERROR, + CANONICAL_OK, + CAPABILITY, + createIngressConformanceHarness, + PROBE_CAPABILITY, + SCHEMA, + stableResponse, +} from '../test-helpers/ingress-conformance-test-harness'; /* eslint-disable @typescript-eslint/no-require-imports */ const brokerDir = path.join(__dirname, '..', '..', 'containers', 'bounded-agent', 'broker'); -const { createServer, createTcpServer, listenOnSocket, listenOnTcp, MAX_CONNECTIONS } = require( +const { MAX_CONNECTIONS } = require( path.join(brokerDir, 'server.js'), ); /* eslint-enable @typescript-eslint/no-require-imports */ @@ -25,132 +30,40 @@ const { createServer, createTcpServer, listenOnSocket, listenOnTcp, MAX_CONNECTI * production surface is widened for these tests. */ -const CAPABILITY = 'a'.repeat(64); -const PROBE_CAPABILITY = 'b'.repeat(64); -const CANONICAL_ERROR = '{"status":"error"}'; -const CANONICAL_OK = '{"status":"ok","result":true}'; -const SCHEMA = Buffer.from('{"type":"boolean"}').toString('base64url'); const MAX_TASK_BYTES = 64 * 1024; -interface Response { - status: number | undefined; - headers: http.IncomingHttpHeaders; - body: string; -} - -function stableResponse(response: Response) { - return { - status: response.status, - body: response.body, - contentType: response.headers['content-type'], - cacheControl: response.headers['cache-control'], - contentLength: response.headers['content-length'], - }; -} - -function request(options: http.RequestOptions, body = 'do the task'): Promise { - return new Promise((resolve, reject) => { - const req = http.request({ - method: 'POST', - path: '/query', - ...options, - headers: { - 'content-type': 'application/octet-stream', - 'x-awf-agent-version': '1', - 'x-awf-repo': 'octo/private', - 'x-awf-schema-b64': SCHEMA, - ...options.headers, - }, - }, (res) => { - const chunks: Buffer[] = []; - res.on('data', (chunk) => chunks.push(chunk)); - res.on('end', () => resolve({ - status: res.statusCode, - headers: res.headers, - body: Buffer.concat(chunks).toString('utf8'), - })); - }); - req.on('error', reject); - req.end(body); - }); -} - describe('bounded-agent ingress conformance', () => { - let root: string; - let unixServer: http.Server; - let tcpServer: http.Server; - let socketPath: string; - let tcpPort: number; - let handled: unknown[]; - const audit = { - failure: jest.fn(), - lifecycle: jest.fn(), - }; + const harness = createIngressConformanceHarness({ + brokerDir, + socketRootPrefix: 'awf-bounded-agent-ingress-test-', + requestHeaders: { + 'x-awf-agent-version': '1', + 'x-awf-repo': 'octo/private', + 'x-awf-schema-b64': SCHEMA, + }, + defaultBody: 'do the task', + }); beforeEach(async () => { - root = fs.mkdtempSync(path.join(os.tmpdir(), 'awf-bounded-agent-ingress-test-')); - socketPath = path.join(root, 'broker.sock'); - handled = []; - const broker = { - handle: (incoming: unknown, respond: (body: string) => void) => { - handled.push(incoming); - respond(incoming === undefined ? CANONICAL_ERROR : CANONICAL_OK); - return Promise.resolve(); - }, - }; - unixServer = createServer({ broker, audit }); - tcpServer = createTcpServer({ - broker, - audit, - capabilities: { query: CAPABILITY, probe: PROBE_CAPABILITY }, - }); - await listenOnSocket(unixServer, { - socketPath, - socketDir: root, - socketUid: process.getuid?.() ?? 0, - socketGid: process.getgid?.() ?? 0, - }, audit); - await listenOnTcp(tcpServer, { tcpPort: 0 }); - tcpPort = (tcpServer.address() as AddressInfo).port; + await harness.start(); }); afterEach(async () => { - await Promise.all([ - new Promise((resolve) => unixServer.close(() => resolve())), - new Promise((resolve) => tcpServer.close(() => resolve())), - ]); - fs.rmSync(root, { recursive: true, force: true }); - jest.clearAllMocks(); + await harness.stop(); }); - const unixRequest = (body?: string) => request({ socketPath }, body); - const tcpRequest = (body?: string, capability = CAPABILITY) => request({ - host: '127.0.0.1', - port: tcpPort, - headers: { 'x-awf-capability': capability }, - }, body); + const { audit, request, tcpRequest, unixRequest } = harness; it('returns byte-identical status, headers, and canonical result bytes across transports', async () => { - const [unix, tcp] = await Promise.all([unixRequest(), tcpRequest()]); - expect(stableResponse(tcp)).toEqual(stableResponse(unix)); - expect(stableResponse(tcp)).toEqual(expect.objectContaining({ - status: 200, - body: CANONICAL_OK, - contentType: 'application/json', - cacheControl: 'no-store', - contentLength: String(Buffer.byteLength(CANONICAL_OK)), - })); - expect(handled).toHaveLength(2); - expect(handled[0]).toEqual(handled[1]); - expect(handled[0]).not.toHaveProperty('capability'); + await harness.expectCanonicalTransportParity(); }); it('collapses missing, wrong, and duplicated authentication to canonical failure bytes', async () => { - const missing = request({ host: '127.0.0.1', port: tcpPort }); + const missing = request({ host: '127.0.0.1', port: harness.tcpPort }); const wrong = tcpRequest(undefined, 'c'.repeat(64)); const duplicated = request({ host: '127.0.0.1', - port: tcpPort, + port: harness.tcpPort, headers: { 'x-awf-capability': [CAPABILITY, CAPABILITY] }, }); const responses = await Promise.all([missing, wrong, duplicated]); @@ -158,17 +71,17 @@ describe('bounded-agent ingress conformance', () => { expect(response.status).toBe(200); expect(response.body).toBe(CANONICAL_ERROR); } - expect(handled).toHaveLength(0); + expect(harness.handled).toHaveLength(0); expect(audit.failure).toHaveBeenCalledWith('transport', 'auth-rejected'); }); it('uses a one-shot probe capability without launching or consuming a request, then permanently retires it', async () => { - const before = handled.length; + const before = harness.handled.length; const first = await tcpRequest('', PROBE_CAPABILITY); const second = await tcpRequest('', PROBE_CAPABILITY); expect(first.body).toBe(CANONICAL_ERROR); expect(second.body).toBe(CANONICAL_ERROR); - expect(handled.length).toBe(before); + expect(harness.handled.length).toBe(before); expect(audit.lifecycle).toHaveBeenCalledWith('sbx-ingress-probe'); expect(audit.lifecycle).toHaveBeenCalledTimes(1); // The second attempt with the same (now-retired) probe capability must be @@ -178,9 +91,9 @@ describe('bounded-agent ingress conformance', () => { it('strips the capability header before handing the request to framing/broker logic', async () => { await tcpRequest(); - expect(handled).toHaveLength(1); - expect(handled[0]).not.toHaveProperty('capability'); - expect(JSON.stringify(handled[0])).not.toContain(CAPABILITY); + expect(harness.handled).toHaveLength(1); + expect(harness.handled[0]).not.toHaveProperty('capability'); + expect(JSON.stringify(harness.handled[0])).not.toContain(CAPABILITY); }); it('keeps oversized and parallel request behavior identical across transports', async () => { @@ -218,13 +131,13 @@ describe('bounded-agent ingress conformance', () => { it('does not dispatch broker work for a request that arrives on an over-limit socket', async () => { const holders = await Promise.all(Array.from({ length: MAX_CONNECTIONS }, () => new Promise((resolve, reject) => { - const socket = net.createConnection({ host: '127.0.0.1', port: tcpPort }, () => resolve(socket)); + const socket = net.createConnection({ host: '127.0.0.1', port: harness.tcpPort }, () => resolve(socket)); socket.on('error', reject); }))); try { const rawResponse = await new Promise((resolve, reject) => { - const socket = net.createConnection({ host: '127.0.0.1', port: tcpPort }, () => { + const socket = net.createConnection({ host: '127.0.0.1', port: harness.tcpPort }, () => { socket.write([ 'POST /query HTTP/1.1', 'Host: 127.0.0.1', @@ -245,7 +158,7 @@ describe('bounded-agent ingress conformance', () => { }); expect(rawResponse).toContain(CANONICAL_ERROR); - expect(handled).toHaveLength(0); + expect(harness.handled).toHaveLength(0); expect(audit.failure).toHaveBeenCalledWith('transport', 'connection-limit'); } finally { for (const socket of holders) socket.destroy(); diff --git a/src/bounded-query/ingress-conformance.test.ts b/src/bounded-query/ingress-conformance.test.ts index 2ebd5e4c3..fb7ab9967 100644 --- a/src/bounded-query/ingress-conformance.test.ts +++ b/src/bounded-query/ingress-conformance.test.ts @@ -1,142 +1,54 @@ -import * as fs from 'fs'; -import * as http from 'http'; import * as net from 'net'; -import * as os from 'os'; import * as path from 'path'; -import type { AddressInfo } from 'net'; +import { + CANONICAL_ERROR, + CANONICAL_OK, + CAPABILITY, + createIngressConformanceHarness, + PROBE_CAPABILITY, + SCHEMA, + stableResponse, +} from '../test-helpers/ingress-conformance-test-harness'; /* eslint-disable @typescript-eslint/no-require-imports */ const brokerDir = path.join(__dirname, '..', '..', 'containers', 'bounded-query', 'broker'); -const { createServer, createTcpServer, listenOnSocket, listenOnTcp, MAX_CONNECTIONS } = require( +const { MAX_CONNECTIONS } = require( path.join(brokerDir, 'server.js'), ); /* eslint-enable @typescript-eslint/no-require-imports */ -const CAPABILITY = 'a'.repeat(64); -const PROBE_CAPABILITY = 'b'.repeat(64); -const CANONICAL_ERROR = '{"status":"error"}'; -const CANONICAL_OK = '{"status":"ok","result":true}'; -const SCHEMA = Buffer.from('{"type":"boolean"}').toString('base64url'); - -interface Response { - status: number | undefined; - headers: http.IncomingHttpHeaders; - body: string; -} - -function stableResponse(response: Response) { - return { - status: response.status, - body: response.body, - contentType: response.headers['content-type'], - cacheControl: response.headers['cache-control'], - contentLength: response.headers['content-length'], - }; -} - -function request(options: http.RequestOptions, body = 'print(True)'): Promise { - return new Promise((resolve, reject) => { - const req = http.request({ - method: 'POST', - path: '/query', - ...options, - headers: { - 'content-type': 'application/octet-stream', - 'x-awf-query-version': '2', - 'x-awf-repo': 'octo/private', - 'x-awf-schema-b64': SCHEMA, - ...options.headers, - }, - }, (res) => { - const chunks: Buffer[] = []; - res.on('data', (chunk) => chunks.push(chunk)); - res.on('end', () => resolve({ - status: res.statusCode, - headers: res.headers, - body: Buffer.concat(chunks).toString('utf8'), - })); - }); - req.on('error', reject); - req.end(body); - }); -} - describe('bounded-query ingress conformance', () => { - let root: string; - let unixServer: http.Server; - let tcpServer: http.Server; - let socketPath: string; - let tcpPort: number; - let handled: unknown[]; - const audit = { - failure: jest.fn(), - lifecycle: jest.fn(), - }; + const harness = createIngressConformanceHarness({ + brokerDir, + socketRootPrefix: 'awf-ingress-test-', + requestHeaders: { + 'x-awf-query-version': '2', + 'x-awf-repo': 'octo/private', + 'x-awf-schema-b64': SCHEMA, + }, + defaultBody: 'print(True)', + }); beforeEach(async () => { - root = fs.mkdtempSync(path.join(os.tmpdir(), 'awf-ingress-test-')); - socketPath = path.join(root, 'broker.sock'); - handled = []; - const broker = { - handle: (incoming: unknown, respond: (body: string) => void) => { - handled.push(incoming); - respond(incoming === undefined ? CANONICAL_ERROR : CANONICAL_OK); - return Promise.resolve(); - }, - }; - unixServer = createServer({ broker, audit }); - tcpServer = createTcpServer({ - broker, - audit, - capabilities: { query: CAPABILITY, probe: PROBE_CAPABILITY }, - }); - await listenOnSocket(unixServer, { - socketPath, - socketDir: root, - socketUid: process.getuid?.() ?? 0, - socketGid: process.getgid?.() ?? 0, - }, audit); - await listenOnTcp(tcpServer, { tcpPort: 0 }); - tcpPort = (tcpServer.address() as AddressInfo).port; + await harness.start(); }); afterEach(async () => { - await Promise.all([ - new Promise((resolve) => unixServer.close(() => resolve())), - new Promise((resolve) => tcpServer.close(() => resolve())), - ]); - fs.rmSync(root, { recursive: true, force: true }); - jest.clearAllMocks(); + await harness.stop(); }); - const unixRequest = (body?: string) => request({ socketPath }, body); - const tcpRequest = (body?: string, capability = CAPABILITY) => request({ - host: '127.0.0.1', - port: tcpPort, - headers: { 'x-awf-capability': capability }, - }, body); + const { audit, request, tcpRequest, unixRequest } = harness; it('returns byte-identical status, headers, and canonical result bytes', async () => { - const [unix, tcp] = await Promise.all([unixRequest(), tcpRequest()]); - expect(stableResponse(tcp)).toEqual(stableResponse(unix)); - expect(stableResponse(tcp)).toEqual(expect.objectContaining({ - status: 200, - body: CANONICAL_OK, - contentType: 'application/json', - cacheControl: 'no-store', - contentLength: String(Buffer.byteLength(CANONICAL_OK)), - })); - expect(handled).toHaveLength(2); - expect(handled[0]).toEqual(handled[1]); - expect(handled[0]).not.toHaveProperty('capability'); + await harness.expectCanonicalTransportParity(); }); it('collapses missing, wrong, and duplicated authentication to canonical failure bytes', async () => { - const missing = request({ host: '127.0.0.1', port: tcpPort }); + const missing = request({ host: '127.0.0.1', port: harness.tcpPort }); const wrong = tcpRequest(undefined, 'c'.repeat(64)); const duplicated = request({ host: '127.0.0.1', - port: tcpPort, + port: harness.tcpPort, headers: { 'x-awf-capability': [CAPABILITY, CAPABILITY] }, }); const responses = await Promise.all([missing, wrong, duplicated]); @@ -144,16 +56,16 @@ describe('bounded-query ingress conformance', () => { expect(response.status).toBe(200); expect(response.body).toBe(CANONICAL_ERROR); } - expect(handled).toHaveLength(0); + expect(harness.handled).toHaveLength(0); }); it('uses a one-shot probe capability without launching or consuming a query request', async () => { - const before = handled.length; + const before = harness.handled.length; const first = await tcpRequest('', PROBE_CAPABILITY); const second = await tcpRequest('', PROBE_CAPABILITY); expect(first.body).toBe(CANONICAL_ERROR); expect(second.body).toBe(CANONICAL_ERROR); - expect(handled.length).toBe(before); + expect(harness.handled.length).toBe(before); expect(audit.lifecycle).toHaveBeenCalledWith('sbx-ingress-probe'); }); @@ -177,13 +89,13 @@ describe('bounded-query ingress conformance', () => { it('does not dispatch broker work for a request that arrives on an over-limit socket', async () => { const holders = await Promise.all(Array.from({ length: MAX_CONNECTIONS }, () => new Promise((resolve, reject) => { - const socket = net.createConnection({ host: '127.0.0.1', port: tcpPort }, () => resolve(socket)); + const socket = net.createConnection({ host: '127.0.0.1', port: harness.tcpPort }, () => resolve(socket)); socket.on('error', reject); }))); try { const rawResponse = await new Promise((resolve, reject) => { - const socket = net.createConnection({ host: '127.0.0.1', port: tcpPort }, () => { + const socket = net.createConnection({ host: '127.0.0.1', port: harness.tcpPort }, () => { socket.write([ 'POST /query HTTP/1.1', 'Host: 127.0.0.1', @@ -204,7 +116,7 @@ describe('bounded-query ingress conformance', () => { }); expect(rawResponse).toContain(CANONICAL_ERROR); - expect(handled).toHaveLength(0); + expect(harness.handled).toHaveLength(0); expect(audit.failure).toHaveBeenCalledWith('transport', 'connection-limit'); } finally { for (const socket of holders) socket.destroy(); diff --git a/src/test-helpers/ingress-conformance-test-harness.ts b/src/test-helpers/ingress-conformance-test-harness.ts new file mode 100644 index 000000000..1eede92a1 --- /dev/null +++ b/src/test-helpers/ingress-conformance-test-harness.ts @@ -0,0 +1,154 @@ +import * as fs from 'fs'; +import * as http from 'http'; +import * as os from 'os'; +import * as path from 'path'; +import type { AddressInfo } from 'net'; + +export const CAPABILITY = 'a'.repeat(64); +export const PROBE_CAPABILITY = 'b'.repeat(64); +export const CANONICAL_ERROR = '{"status":"error"}'; +export const CANONICAL_OK = '{"status":"ok","result":true}'; +export const SCHEMA = Buffer.from('{"type":"boolean"}').toString('base64url'); + +interface Response { + status: number | undefined; + headers: http.IncomingHttpHeaders; + body: string; +} + +interface BrokerServerModule { + createServer: (options: object) => http.Server; + createTcpServer: (options: object) => http.Server; + listenOnSocket: (server: http.Server, options: object, audit: object) => Promise; + listenOnTcp: (server: http.Server, options: object) => Promise; +} + +interface IngressConformanceHarnessOptions { + brokerDir: string; + socketRootPrefix: string; + requestHeaders: http.OutgoingHttpHeaders; + defaultBody: string; +} + +export function stableResponse(response: Response) { + return { + status: response.status, + body: response.body, + contentType: response.headers['content-type'], + cacheControl: response.headers['cache-control'], + contentLength: response.headers['content-length'], + }; +} + +export function createIngressConformanceHarness(options: IngressConformanceHarnessOptions) { + /* eslint-disable @typescript-eslint/no-require-imports, security/detect-non-literal-require */ + const server: BrokerServerModule = require(path.join(options.brokerDir, 'server.js')); + /* eslint-enable @typescript-eslint/no-require-imports, security/detect-non-literal-require */ + let root: string; + let unixServer: http.Server; + let tcpServer: http.Server; + let socketPath: string; + let tcpPort: number; + let handled: unknown[]; + const audit = { + failure: jest.fn(), + lifecycle: jest.fn(), + }; + + function request(requestOptions: http.RequestOptions, body = options.defaultBody): Promise { + return new Promise((resolve, reject) => { + const req = http.request({ + method: 'POST', + path: '/query', + ...requestOptions, + headers: { + 'content-type': 'application/octet-stream', + ...options.requestHeaders, + ...requestOptions.headers, + }, + }, (res) => { + const chunks: Buffer[] = []; + res.on('data', (chunk) => chunks.push(chunk)); + res.on('end', () => resolve({ + status: res.statusCode, + headers: res.headers, + body: Buffer.concat(chunks).toString('utf8'), + })); + }); + req.on('error', reject); + req.end(body); + }); + } + + return { + audit, + get handled() { + return handled; + }, + get tcpPort() { + return tcpPort; + }, + async start() { + root = fs.mkdtempSync(path.join(os.tmpdir(), options.socketRootPrefix)); + socketPath = path.join(root, 'broker.sock'); + handled = []; + const broker = { + handle: (incoming: unknown, respond: (body: string) => void) => { + handled.push(incoming); + respond(incoming === undefined ? CANONICAL_ERROR : CANONICAL_OK); + return Promise.resolve(); + }, + }; + unixServer = server.createServer({ broker, audit }); + tcpServer = server.createTcpServer({ + broker, + audit, + capabilities: { query: CAPABILITY, probe: PROBE_CAPABILITY }, + }); + await server.listenOnSocket(unixServer, { + socketPath, + socketDir: root, + socketUid: process.getuid?.() ?? 0, + socketGid: process.getgid?.() ?? 0, + }, audit); + await server.listenOnTcp(tcpServer, { tcpPort: 0 }); + tcpPort = (tcpServer.address() as AddressInfo).port; + }, + async stop() { + await Promise.all([ + new Promise((resolve) => unixServer.close(() => resolve())), + new Promise((resolve) => tcpServer.close(() => resolve())), + ]); + fs.rmSync(root, { recursive: true, force: true }); + jest.clearAllMocks(); + }, + request, + unixRequest: (body?: string) => request({ socketPath }, body), + tcpRequest: (body?: string, capability = CAPABILITY) => request({ + host: '127.0.0.1', + port: tcpPort, + headers: { 'x-awf-capability': capability }, + }, body), + async expectCanonicalTransportParity() { + const [unix, tcp] = await Promise.all([ + request({ socketPath }), + request({ + host: '127.0.0.1', + port: tcpPort, + headers: { 'x-awf-capability': CAPABILITY }, + }), + ]); + expect(stableResponse(tcp)).toEqual(stableResponse(unix)); + expect(stableResponse(tcp)).toEqual(expect.objectContaining({ + status: 200, + body: CANONICAL_OK, + contentType: 'application/json', + cacheControl: 'no-store', + contentLength: String(Buffer.byteLength(CANONICAL_OK)), + })); + expect(handled).toHaveLength(2); + expect(handled[0]).toEqual(handled[1]); + expect(handled[0]).not.toHaveProperty('capability'); + }, + }; +}