From 0f7005e0a74b0fb6fe769951f0a9e11e4b2bc0aa Mon Sep 17 00:00:00 2001 From: "Nikhil Chitlur Navakiran (from Dev Box)" Date: Wed, 28 Oct 2026 09:48:17 -0600 Subject: [PATCH 1/2] feat(a365): port tool transfer and tenant-isolated delivery parity Capture complete tool URLs and explicit transfer targets, isolate durable backoff and replay by tenant, and preserve empty nested metadata. Include focused regressions and the JavaScript parity documentation. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: adab09b1-cb4f-4d83-ad92-31d3462610ec --- A365_DOCUMENTATION.md | 72 +++++- CHANGELOG.md | 4 + src/a365/constants.ts | 20 ++ src/a365/contracts.ts | 43 +++- .../durable/DurableDeliveryManager.ts | 127 ++++++---- src/a365/exporter/durable/DurableRecord.ts | 1 + src/a365/exporter/durable/PersistentStore.ts | 5 +- src/a365/exporter/durable/TransmissionGate.ts | 4 + .../durable/TransmissionGateRegistry.ts | 91 +++++++ src/a365/index.ts | 3 + src/a365/message-utils.ts | 5 +- src/a365/scopes/ExecuteToolScope.ts | 36 +++ src/index.ts | 3 + .../unit/a365/durableDeliveryManager.test.ts | 235 +++++++++++++----- test/internal/unit/a365/durableRecord.test.ts | 11 + test/internal/unit/a365/messageUtils.test.ts | 16 ++ .../unit/a365/persistentStore.test.ts | 63 +++++ test/internal/unit/a365/scopes.test.ts | 122 +++++++++ .../a365/transmissionGateRegistry.test.ts | 102 ++++++++ 19 files changed, 843 insertions(+), 120 deletions(-) create mode 100644 src/a365/exporter/durable/TransmissionGateRegistry.ts create mode 100644 test/internal/unit/a365/transmissionGateRegistry.test.ts diff --git a/A365_DOCUMENTATION.md b/A365_DOCUMENTATION.md index 23381c1e9..c0c0155f0 100644 --- a/A365_DOCUMENTATION.md +++ b/A365_DOCUMENTATION.md @@ -126,14 +126,62 @@ invokeScope.dispose(); `ExecuteToolScope` serializes arguments to `gen_ai.tool.call.arguments` and results to `gen_ai.tool.call.result` as JSON span attributes, so they may contain sensitive data. -Use `extension_data` for provider-specific fields on any typed ExecuteTool model. Non-empty -extension data is emitted under the model's `metadata` JSON property. Metadata keys remain +Use `extension_data` for provider-specific fields on any typed ExecuteTool model. Explicitly +provided extension data, including `{}`, is emitted under the model's `metadata` JSON property; +absent or null extension data is omitted. Metadata keys remain isolated from declared schema fields, so an `extension_data.action` or `extension_data.schema_version` value cannot replace the typed `action` or `schema_version`. Typed payloads that contain invalid enum tokens, non-finite numbers, unsupported values, or reference cycles are replaced with `{"serialization_error":"Failed to serialize execute tool payload."}`. +### Tool endpoints and transfers + +Supply `endpoint.url` on `ToolCallDetails` to record the complete URL as `url.full`, +including its path, query, and fragment. Existing `endpoint.host` and non-default +`endpoint.port` still populate `server.address` and `server.port`. Host-only endpoints +do not synthesize `url.full`. URLs may contain sensitive data; sanitize them before recording. + +Use explicit `transferDetails` for handoffs; ordinary tool calls emit no transfer attributes. + +```typescript +import { ExecuteToolScope, TransferMode, TransferTargetType } from "@microsoft/opentelemetry"; +import type { ToolCallDetails } from "@microsoft/opentelemetry"; + +const details: ToolCallDetails = { + toolName: "handoff", + endpoint: { + host: "agents.example.com", + url: "https://agents.example.com/support", + }, + transferDetails: { + mode: TransferMode.PASS_CONTROL, + targetType: TransferTargetType.AGENT, + targetAgentDetails: { agentId: "support-agent", agentName: "Support" }, + }, +}; +const scope = ExecuteToolScope.start( + { conversationId: "conv-123" }, + details, + { agentId: "source-agent", tenantId: "tenant-1" }, +); +scope.dispose(); +``` + +`TransferMode` supports `RETURN_TO_CALLER` (`return_to_caller`) and `PASS_CONTROL` +(`pass_control`). `TransferTargetType` supports `AGENT`, `HUMAN`, and `WORKFLOW`; +omitting `targetType` defaults to `agent`. Target identity is optional for agent transfers +and forbidden for human/workflow transfers. Invalid enum values or incompatible target +details throw before a span is created. + +Transfers emit `microsoft.a365.transfer.mode` and `microsoft.a365.transfer.target.type`. +Provided agent ID, name, blueprint ID, platform ID, and version are recorded under +`microsoft.a365.transfer.target.agent.*` (`id`, `name`, `blueprint.id`, `platform.id`, +`version`). Other target identity fields are not emitted, and the source agent and tenant +attributes remain unchanged. + +### Request and response attributes + `InvokeAgentScope`, `InferenceScope`, and `ExecuteToolScope` accept `request.sessionId`. When you provide it, those scopes write `microsoft.session.id` directly on the created span instead of relying on later baggage enrichment. `OutputScope` does not currently @@ -296,7 +344,7 @@ network-only delivery. It applies only to the A365 HTTP exporter, so set | `maxStorageBytes` | `50 * 1024 * 1024` | Bounds pending, quarantined, active leased, and non-stale temporary records within the current `app-` partition only | | `maxRecordAgeMilliseconds` | `2 * 24 * 60 * 60 * 1000` | Expired records are pruned before capacity eviction, within the current `app-` partition only | | `replayIntervalMilliseconds` | `2 * 60 * 1000` | Scheduled replay cadence | -| `maxReplayBatchSize` | `10` | Maximum records claimed per replay pass | +| `maxReplayBatchSize` | `10` | Maximum attempted records per replay pass; backed-off records do not consume this limit. At most 10 times this many records are claimed | | `leaseDurationMilliseconds` | `2 * 60 * 1000` | Reclaims stale replay leases | | `shutdownTimeoutMilliseconds` | `10_000` | Shared shutdown budget for accepted live exports and admitted durable handoff completion | | `tokenResolutionTimeoutMilliseconds` | `30_000` | Timeout per replay token-resolution attempt | @@ -333,20 +381,28 @@ network-only delivery. It applies only to the A365 HTTP exporter, so set - Delivery is at-least-once. A retryable request can be replayed after a crash, timeout, or shutdown race, so downstream consumers must tolerate duplicates. - HTTP 401, 408, 429, and 5xx responses plus transport failures are retryable. Live delivery - persists them, and replay releases the claim while honoring the shared transmission gate. + persists them, and replay releases the claim while honoring that tenant's transmission gate. - Replay and `forceFlush()` resolve a fresh token for each send attempt and use the exporter's current cluster/domain routing; durable files do not store bearer tokens or authoritative route metadata. - If token resolution returns no token, throws, or times out, live delivery attempts to persist the - record for replay and replay releases the claim without extending the shared transmission backoff. + record for replay and replay releases the claim without extending tenant transmission backoff. - Storage is bounded by both age and capacity, scoped to the current process's own `app-` partition (see above). The SDK sweeps stale temporary files, prunes expired records, then evicts the oldest remaining pending or quarantined record until the new record fits within `maxStorageBytes`; active temporary and leased files count against that bound and are not evicted. `maxStorageBytes` is therefore a per-partition, not a global, limit. -- Live delivery and replay share one `Retry-After` / exponential-backoff transmission gate. A - retryable response pauses both immediate sends and replay probes until the gate reopens. The - effective delay is capped at one hour. +- Live delivery and replay share a `Retry-After` / exponential-backoff transmission gate per + tenant, across that tenant's agents. A retryable response pauses only that tenant's immediate + sends and replay probes; other tenants continue. The effective delay is capped at one hour. + Replay skips released records within the same pass and scans past backed-off tenants within + the bounded claim budget (not a limit on directory listings or preflight reads). Excluded + records are skipped before leasing; post-claim content is still validated. Malformed records, + including blank tenant IDs, are quarantined. Invalid replay endpoints, unexpected send + exceptions, or failed claim releases stop the pass. +- Tenant gates are retained in a registry with a soft capacity of 256 and a 15-minute idle + timeout. Inactive gates are evicted first; active sends and unexpired backoff are never evicted, + so the registry may temporarily exceed its capacity. - If records must survive container restarts, rescheduling, or host restarts, point `storageDirectory` at a protected persistent volume. Ephemeral storage can be lost when a container is replaced and still counts against container ephemeral-storage limits. diff --git a/CHANGELOG.md b/CHANGELOG.md index bf9cd665e..4681a2791 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,11 +3,15 @@ ## [Unreleased] ### Bugs Fixed +- Isolate Agent365 durable delivery backoff by tenant, shared across that tenant's agents and live/replay paths. Continue replay past retryable failures and backed-off tenants without spending the attempt limit on deferred records; bound replay scans and preserve active backoff during registry eviction. +- Skip excluded durable records before leasing and quarantine blank-tenant records before replay, retaining post-claim content validation. +- Preserve explicitly empty `extension_data` as nested `metadata: {}` in typed ExecuteTool payloads, including nested models, while continuing to omit absent/null metadata. - Update OpenAI Agents to 0.15.0, Vitest and its coverage provider to 4.1.12, MSAL Node to 7.1.1, and vulnerable transitive dependencies across the distro and samples. OpenAI Agents 0.15.0 removes the vulnerable MCP/Express dependency chain. - Require Node.js 22.23.4+, 24.22.1+, or 26.12.1+ on their respective release lines for security fixes. - **Breaking change on Unix (Linux/macOS):** Agent365 offline buffering/replay now requires an explicitly configured, existing, absolute `durableDelivery.storageDirectory`. No automatic Unix temporary-directory fallback or root creation. Missing/invalid roots are logged and disable only disk buffering/replay; online export continues. Customers provision and maintain ownership, permissions (normally a private service-account-owned `0700` root), ACLs, and path-replacement protection; the SDK does not inspect or repair Unix access controls. Windows behavior, explicit-root cache identities, record compatibility, and retention limits are unchanged. Old temporary backlogs are not automatically migrated; see [storage provisioning and migration](A365_DOCUMENTATION.md#storage-provisioning-and-migration). ### Features Added +- Add optional full tool endpoint URLs (`endpoint.url` -> `url.full`) and typed execute-tool transfer metadata for agent, human, and workflow targets. Export transfer enums and contracts, validate incompatible targets before creating spans, and preserve source agent identity. - Add privacy-safe Model Observation attributes to `InferenceScope`, including the canonical served response model and cache read/write input-token counts. Attributes can be supplied when the scope starts or recorded after the response is available; inference operations accept enum values or canonical strings and use canonical OpenTelemetry names for both operation attributes and span names. Cache-token semantic-convention constants are exported. [#261](https://github.com/microsoft/opentelemetry-distro-javascript/pull/261) ### Other Changes diff --git a/src/a365/constants.ts b/src/a365/constants.ts index a57b3c0a9..fd7d2c068 100644 --- a/src/a365/constants.ts +++ b/src/a365/constants.ts @@ -42,6 +42,26 @@ export class OpenTelemetryConstants { public static readonly SERVER_ADDRESS_KEY = "server.address"; /** Attribute key for the destination server port (`server.port`). */ public static readonly SERVER_PORT_KEY = "server.port"; + /** Attribute key for the complete tool endpoint URL (`url.full`). */ + public static readonly URL_FULL_KEY = "url.full"; + /** How control passes during an execute-tool transfer. */ + public static readonly TRANSFER_MODE_KEY = "microsoft.a365.transfer.mode"; + /** Kind of target receiving an execute-tool transfer. */ + public static readonly TRANSFER_TARGET_TYPE_KEY = "microsoft.a365.transfer.target.type"; + /** Target agent identifier for an execute-tool transfer. */ + public static readonly TRANSFER_TARGET_AGENT_ID_KEY = "microsoft.a365.transfer.target.agent.id"; + /** Target agent display name for an execute-tool transfer. */ + public static readonly TRANSFER_TARGET_AGENT_NAME_KEY = + "microsoft.a365.transfer.target.agent.name"; + /** Target agent blueprint identifier for an execute-tool transfer. */ + public static readonly TRANSFER_TARGET_AGENT_BLUEPRINT_ID_KEY = + "microsoft.a365.transfer.target.agent.blueprint.id"; + /** Target agent platform identifier for an execute-tool transfer. */ + public static readonly TRANSFER_TARGET_AGENT_PLATFORM_ID_KEY = + "microsoft.a365.transfer.target.agent.platform.id"; + /** Target agent version for an execute-tool transfer. */ + public static readonly TRANSFER_TARGET_AGENT_VERSION_KEY = + "microsoft.a365.transfer.target.agent.version"; /** Azure resource provider namespace value for Cognitive Services. */ public static readonly AZURE_RP_NAMESPACE_VALUE = "Microsoft.CognitiveServices"; diff --git a/src/a365/contracts.ts b/src/a365/contracts.ts index 5058d3252..f4e73743e 100644 --- a/src/a365/contracts.ts +++ b/src/a365/contracts.ts @@ -383,7 +383,46 @@ export interface ServiceEndpoint { port?: number; /** Protocol scheme of the endpoint (e.g. `https`). */ protocol?: string; -} + /** Complete endpoint URL, including path and query, recorded as `url.full` on tool spans. */ + url?: string; +} + +/** How control passes during an execute-tool transfer. */ +export enum TransferMode { + /** The source agent resumes after the target finishes. */ + RETURN_TO_CALLER = "return_to_caller", + /** The target takes over the remaining work. */ + PASS_CONTROL = "pass_control", +} + +/** Kind of target receiving an execute-tool transfer. */ +export enum TransferTargetType { + /** Transfer to another agent. */ + AGENT = "agent", + /** Transfer to a human. */ + HUMAN = "human", + /** Transfer to a workflow. */ + WORKFLOW = "workflow", +} + +/** Explicit transfer metadata. Target agent details are only valid for agent targets. */ +export type TransferDetails = + | { + /** How control passes to the target. */ + mode: TransferMode; + /** Target kind. Defaults to agent when omitted. */ + targetType?: TransferTargetType.AGENT; + /** Target identity; only ID, name, blueprint, platform, and version are emitted. */ + targetAgentDetails?: AgentDetails; + } + | { + /** How control passes to the target. */ + mode: TransferMode; + /** Non-agent target kind. */ + targetType: TransferTargetType.HUMAN | TransferTargetType.WORKFLOW; + /** Agent identity cannot be supplied for a non-agent target. */ + targetAgentDetails?: never; + }; // --------------------------------------------------------------------------- // Scope detail types @@ -455,6 +494,8 @@ export interface ToolCallDetails { toolType?: string; /** Endpoint the tool is hosted on. */ endpoint?: ServiceEndpoint; + /** Optional explicit transfer performed by this tool call. */ + transferDetails?: TransferDetails; } /** Details for an inference call. */ diff --git a/src/a365/exporter/durable/DurableDeliveryManager.ts b/src/a365/exporter/durable/DurableDeliveryManager.ts index d71417b9d..3df08042c 100644 --- a/src/a365/exporter/durable/DurableDeliveryManager.ts +++ b/src/a365/exporter/durable/DurableDeliveryManager.ts @@ -6,7 +6,8 @@ import type { ILogger } from "../../logging.js"; import { ResolvedDurableDeliveryOptions } from "./DurableDeliveryOptions.js"; import type { DurableRecordV1 } from "./DurableRecord.js"; import { PersistentStore, type ClaimedRecord } from "./PersistentStore.js"; -import { TransmissionGate, type TransmissionPermit } from "./TransmissionGate.js"; +import type { TransmissionGate, TransmissionPermit } from "./TransmissionGate.js"; +import { TransmissionGateRegistry } from "./TransmissionGateRegistry.js"; export type DeliveryAttempt = | { kind: "success"; correlationId: string } @@ -30,15 +31,15 @@ type GateDecision = /** * Outcome of processing one claimed replay record: * - "advance": completed (success/permanent); safe to claim the next record. - * - "exclude": token was unavailable; the claim was released and should be - * skipped on later claims in this same pass so later records are not - * starved by repeatedly reclaiming it. - * - "stop": a retryable failure or backed-off gate; stop this replay pass. + * - "exclude": token unavailable or retryable response; count the attempt but + * skip this released record on later claims in the same pass. + * - "deferred": tenant gate is backed off; skip without spending an attempt. + * - "stop": unknown fault or global misconfiguration; stop this replay pass. */ -type ReplayClaimOutcome = "advance" | "exclude" | "stop"; +type ReplayClaimOutcome = "advance" | "exclude" | "deferred" | "stop"; export class DurableDeliveryManager { - private readonly gate = new TransmissionGate(); + private readonly gates = new TransmissionGateRegistry(); private readonly abortController = new AbortController(); private readonly active = new Set>(); private replayTimer?: ReturnType; @@ -186,8 +187,14 @@ export class DurableDeliveryManager { private async runReplayPassInternal(): Promise { const skippedRecordIds = new Set(); - - for (let index = 0; index < this.options.maxReplayBatchSize; index++) { + const scanBudget = Math.min(Number.MAX_SAFE_INTEGER, this.options.maxReplayBatchSize * 10); + let attempted = 0; + + for ( + let scanned = 0; + attempted < this.options.maxReplayBatchSize && scanned < scanBudget; + scanned++ + ) { if (this.closed || this.replayStopped) { return; } @@ -208,7 +215,10 @@ export class DurableDeliveryManager { if (outcome === "stop") { return; } - if (outcome === "exclude") { + if (outcome !== "deferred") { + attempted++; + } + if (outcome === "exclude" || outcome === "deferred") { skippedRecordIds.add(claim.record.id); } } @@ -236,7 +246,9 @@ export class DurableDeliveryManager { ); } - await this.releaseClaim(claim); + if (!(await this.releaseClaim(claim))) { + return "stop"; + } // Exclude this record from later claims in the same pass instead of // stopping outright, so unrelated records that sort after it are not // starved by repeatedly reclaiming the same released record. @@ -249,8 +261,13 @@ export class DurableDeliveryManager { ); } - await this.releaseClaim(claim); - return "stop"; + if (!(await this.releaseClaim(claim))) { + return "stop"; + } + if (decision.kind === "deferred") { + return "deferred"; + } + return decision.kind === "retryable" && decision.error === undefined ? "exclude" : "stop"; } private async persist(record: DurableRecordV1): Promise { @@ -263,11 +280,13 @@ export class DurableDeliveryManager { } } - private async releaseClaim(claim: ClaimedRecord): Promise { + private async releaseClaim(claim: ClaimedRecord): Promise { try { await this.store.release(claim); + return true; } catch (error) { this.logger.error("[DurableDeliveryManager] Failed to release durable record", error); + return false; } } @@ -318,57 +337,67 @@ export class DurableDeliveryManager { record: DurableRecordV1, mode: "live" | "replay", ): Promise { - if (!this.gate.canAcquire()) { - return { kind: "deferred" }; - } + const lease = this.gates.acquire(record.tenantId); + const gate = lease.gate; + try { + if (!gate.canAcquire()) { + return { kind: "deferred" }; + } - if (mode === "replay") { - const decision = await this.validateReplay(record); - if (decision) { - return decision; + if (mode === "replay") { + const decision = await this.validateReplay(record); + if (decision) { + return decision; + } } - } - let token: string | null; - try { - token = await this.withTimeout( - this.dependencies.resolveToken(record), - this.options.tokenResolutionTimeoutMilliseconds, - "token resolution", - ); - } catch (error) { - return { kind: "tokenUnavailable", error }; - } + let token: string | null; + try { + token = await this.withTimeout( + this.dependencies.resolveToken(record), + this.options.tokenResolutionTimeoutMilliseconds, + "token resolution", + ); + } catch (error) { + return { kind: "tokenUnavailable", error }; + } - if (!token) { - return { kind: "tokenUnavailable" }; - } + if (!token) { + return { kind: "tokenUnavailable" }; + } - const permit = this.gate.acquire(); - if (!permit) { - return { kind: "deferred" }; - } + const permit = gate.acquire(); + if (!permit) { + return { kind: "deferred" }; + } - try { - const attempt = await this.dependencies.send(record, token, this.abortController.signal); - return this.classifyAttempt(permit, attempt); - } catch (error) { - this.gate.recordRetryableFailure(permit); - return { kind: "retryable", error }; + try { + const attempt = await this.dependencies.send(record, token, this.abortController.signal); + return this.classifyAttempt(gate, permit, attempt); + } catch (error) { + gate.recordRetryableFailure(permit); + return { kind: "retryable", error }; + } + } finally { + lease.release(); } } - private classifyAttempt(permit: TransmissionPermit, attempt: DeliveryAttempt): GateDecision { + private classifyAttempt( + gate: TransmissionGate, + permit: TransmissionPermit, + attempt: DeliveryAttempt, + ): GateDecision { if (attempt.kind === "success") { - this.gate.recordSuccess(permit); + gate.recordSuccess(permit); return { kind: "success" }; } if (attempt.kind === "permanent") { - this.gate.recordSuccess(permit); + gate.recordSuccess(permit); return { kind: "permanent" }; } - this.gate.recordRetryableFailure(permit, attempt.retryAfterMs); + gate.recordRetryableFailure(permit, attempt.retryAfterMs); return { kind: "retryable" }; } diff --git a/src/a365/exporter/durable/DurableRecord.ts b/src/a365/exporter/durable/DurableRecord.ts index 9cede8436..35b5f7963 100644 --- a/src/a365/exporter/durable/DurableRecord.ts +++ b/src/a365/exporter/durable/DurableRecord.ts @@ -74,6 +74,7 @@ export function parseDurableRecord(text: string): DurableRecordV1 { !isSafeDurableRecordId(value.id) || !isValidCreatedAt(value.createdAt) || typeof value.tenantId !== "string" || + value.tenantId.trim().length === 0 || typeof value.agentId !== "string" || typeof value.useS2SEndpoint !== "boolean" || typeof value.body !== "string" || diff --git a/src/a365/exporter/durable/PersistentStore.ts b/src/a365/exporter/durable/PersistentStore.ts index 1fc22d322..b729ebb1a 100644 --- a/src/a365/exporter/durable/PersistentStore.ts +++ b/src/a365/exporter/durable/PersistentStore.ts @@ -162,7 +162,10 @@ export class PersistentStore { } try { - parseDurableRecord(preflightText); + const preflightRecord = parseDurableRecord(preflightText); + if (excludeRecordIds?.has(preflightRecord.id)) { + continue; + } } catch (error) { await this.quarantinePendingFile(pendingPath); this.logger.warn("[PersistentStore] Quarantined malformed pending durable record", error); diff --git a/src/a365/exporter/durable/TransmissionGate.ts b/src/a365/exporter/durable/TransmissionGate.ts index e271d5603..ae1ed3add 100644 --- a/src/a365/exporter/durable/TransmissionGate.ts +++ b/src/a365/exporter/durable/TransmissionGate.ts @@ -22,6 +22,10 @@ export class TransmissionGate { } = {}, ) {} + get hasPendingBackoff(): boolean { + return this.now() < this.blockedUntil; + } + canAcquire(): boolean { const currentTime = this.now(); if (currentTime < this.blockedUntil) return false; diff --git a/src/a365/exporter/durable/TransmissionGateRegistry.ts b/src/a365/exporter/durable/TransmissionGateRegistry.ts new file mode 100644 index 000000000..7a8c279b5 --- /dev/null +++ b/src/a365/exporter/durable/TransmissionGateRegistry.ts @@ -0,0 +1,91 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +import { TransmissionGate } from "./TransmissionGate.js"; + +interface GateEntry { + gate: TransmissionGate; + activeLeases: number; + lastAccess: number; +} + +/** Shares backoff by tenant without evicting active sends or pending backoff. */ +export class TransmissionGateRegistry { + private readonly entries = new Map(); + private readonly capacity: number; + private readonly idleTimeoutMilliseconds: number; + private readonly now: () => number; + + constructor( + options: { + capacity?: number; + idleTimeoutMilliseconds?: number; + now?: () => number; + } = {}, + ) { + this.capacity = options.capacity ?? 256; + this.idleTimeoutMilliseconds = options.idleTimeoutMilliseconds ?? 15 * 60 * 1000; + this.now = options.now ?? Date.now; + if (!Number.isSafeInteger(this.capacity) || this.capacity < 1) { + throw new RangeError("Transmission gate registry capacity must be a positive integer"); + } + if (!Number.isFinite(this.idleTimeoutMilliseconds) || this.idleTimeoutMilliseconds <= 0) { + throw new RangeError("Transmission gate registry idle timeout must be positive"); + } + } + + acquire(tenantId: string): { gate: TransmissionGate; release(): void } { + if (!tenantId?.trim()) { + throw new TypeError("Transmission gate registry requires a tenant identifier"); + } + + const now = this.now(); + this.trim(now); + let entry = this.entries.get(tenantId); + if (!entry) { + entry = { + gate: new TransmissionGate({ now: this.now }), + activeLeases: 0, + lastAccess: now, + }; + this.entries.set(tenantId, entry); + } + const acquired = entry; + acquired.activeLeases++; + acquired.lastAccess = now; + this.trim(now); + let released = false; + + return { + gate: acquired.gate, + release: () => { + if (released) return; + released = true; + acquired.activeLeases--; + acquired.lastAccess = this.now(); + this.trim(acquired.lastAccess); + }, + }; + } + + private trim(now: number): void { + const inactive: [string, GateEntry][] = []; + for (const [tenantId, entry] of this.entries) { + if (entry.activeLeases > 0 || entry.gate.hasPendingBackoff) continue; + if (now - entry.lastAccess >= this.idleTimeoutMilliseconds) { + this.entries.delete(tenantId); + } else { + inactive.push([tenantId, entry]); + } + } + + inactive.sort( + ([leftId, left], [rightId, right]) => + left.lastAccess - right.lastAccess || (leftId < rightId ? -1 : leftId > rightId ? 1 : 0), + ); + for (const [tenantId] of inactive) { + if (this.entries.size <= this.capacity) break; + this.entries.delete(tenantId); + } + } +} diff --git a/src/a365/index.ts b/src/a365/index.ts index 8ea2ac8fd..5a1c32f4a 100644 --- a/src/a365/index.ts +++ b/src/a365/index.ts @@ -37,6 +37,8 @@ export { Modality, InvocationRole, InferenceOperationType, + TransferMode, + TransferTargetType, DEFAULT_FINISH_REASON, ToolCallAction, ToolCallOutcomeStatus, @@ -81,6 +83,7 @@ export type { GenAiResponseParameters, InvokeAgentScopeDetails, ToolCallDetails, + TransferDetails, InferenceDetails, InferenceResponse, OutputResponse, diff --git a/src/a365/message-utils.ts b/src/a365/message-utils.ts index d082c1695..54c6ce37e 100644 --- a/src/a365/message-utils.ts +++ b/src/a365/message-utils.ts @@ -393,10 +393,7 @@ function withMetadata( if (!isPlainRecord(extensionData)) { throw new TypeError("Execute tool extension data must be a plain object."); } - const metadata = toJsonRecord(extensionData, stack); - if (Object.keys(metadata).length > 0) { - serialized.metadata = metadata; - } + serialized.metadata = toJsonRecord(extensionData, stack); } return serialized; } diff --git a/src/a365/scopes/ExecuteToolScope.ts b/src/a365/scopes/ExecuteToolScope.ts index 5774e23a1..648eab821 100644 --- a/src/a365/scopes/ExecuteToolScope.ts +++ b/src/a365/scopes/ExecuteToolScope.ts @@ -5,6 +5,7 @@ import { SpanKind } from "@opentelemetry/api"; import { OpenTelemetryScope } from "./OpenTelemetryScope.js"; import { OpenTelemetryConstants } from "../constants.js"; import { safeSerializeToJson, serializeToolPayload } from "../message-utils.js"; +import { TransferMode, TransferTargetType } from "../contracts.js"; import type { ToolCallDetails, AgentDetails, @@ -48,6 +49,20 @@ export class ExecuteToolScope extends OpenTelemetryScope { throw new Error("ExecuteToolScope: tenantId is required on agentDetails"); } + const transfer = details.transferDetails; + const targetType = transfer?.targetType ?? TransferTargetType.AGENT; + if (transfer) { + if (!Object.values(TransferMode).includes(transfer.mode)) { + throw new RangeError("ExecuteToolScope: invalid transfer mode"); + } + if (!Object.values(TransferTargetType).includes(targetType)) { + throw new RangeError("ExecuteToolScope: invalid transfer target type"); + } + if (targetType !== TransferTargetType.AGENT && transfer.targetAgentDetails != null) { + throw new TypeError("ExecuteToolScope: target agent details require an agent transfer"); + } + } + const resolvedSpanDetails: SpanDetails = { spanKind: SpanKind.INTERNAL, ...spanDetails, @@ -75,8 +90,29 @@ export class ExecuteToolScope extends OpenTelemetryScope { this.setTagMaybe(OpenTelemetryConstants.GEN_AI_TOOL_CALL_ID_KEY, toolCallId); this.setTagMaybe(OpenTelemetryConstants.GEN_AI_TOOL_DESCRIPTION_KEY, description); + if (transfer) { + this.setTagMaybe(OpenTelemetryConstants.TRANSFER_MODE_KEY, transfer.mode); + this.setTagMaybe(OpenTelemetryConstants.TRANSFER_TARGET_TYPE_KEY, targetType); + const target = transfer.targetAgentDetails; + this.setTagMaybe(OpenTelemetryConstants.TRANSFER_TARGET_AGENT_ID_KEY, target?.agentId); + this.setTagMaybe(OpenTelemetryConstants.TRANSFER_TARGET_AGENT_NAME_KEY, target?.agentName); + this.setTagMaybe( + OpenTelemetryConstants.TRANSFER_TARGET_AGENT_BLUEPRINT_ID_KEY, + target?.agentBlueprintId, + ); + this.setTagMaybe( + OpenTelemetryConstants.TRANSFER_TARGET_AGENT_PLATFORM_ID_KEY, + target?.platformId, + ); + this.setTagMaybe( + OpenTelemetryConstants.TRANSFER_TARGET_AGENT_VERSION_KEY, + target?.agentVersion, + ); + } + if (endpoint) { this.setTagMaybe(OpenTelemetryConstants.SERVER_ADDRESS_KEY, endpoint.host); + this.setTagMaybe(OpenTelemetryConstants.URL_FULL_KEY, endpoint.url); if (endpoint.port && endpoint.port !== 443) { this.setTagMaybe(OpenTelemetryConstants.SERVER_PORT_KEY, endpoint.port); } diff --git a/src/index.ts b/src/index.ts index 61e3705df..4f11faf3f 100644 --- a/src/index.ts +++ b/src/index.ts @@ -49,6 +49,8 @@ export { ToolCallAction, ToolCallOutcomeStatus, ToolPolicyDecision, + TransferMode, + TransferTargetType, ExecuteToolCallArguments, ExecuteToolCallResult, GuardrailDecisionType, @@ -80,6 +82,7 @@ export type { GenAiResponseParameters, InvokeAgentScopeDetails, ToolCallDetails, + TransferDetails, InferenceDetails, InferenceResponse, OutputResponse, diff --git a/test/internal/unit/a365/durableDeliveryManager.test.ts b/test/internal/unit/a365/durableDeliveryManager.test.ts index ff86f463d..95405927a 100644 --- a/test/internal/unit/a365/durableDeliveryManager.test.ts +++ b/test/internal/unit/a365/durableDeliveryManager.test.ts @@ -16,6 +16,7 @@ import { import type { Agent365DurableDeliveryOptions, ClaimedRecord, + ClaimBatchOptions, DeliveryAttempt, DurableDeliveryDependencies, DurableRecordV1, @@ -70,6 +71,94 @@ describe("DurableDeliveryManager", () => { assert.strictEqual(persist.mock.calls.length, 2); }); + it("isolates live backoff by tenant while sharing it across that tenant's agents", async () => { + const { manager, persist, resolveToken, send } = createManager(); + send.mockResolvedValueOnce({ + kind: "retryable", + correlationId: "tenant-a-throttled", + retryAfterMs: 60_000, + }); + + await manager.deliver(makeRecord({ id: "a1", tenantId: "a", agentId: "one" })); + await manager.deliver(makeRecord({ id: "a2", tenantId: "a", agentId: "two" })); + await manager.deliver(makeRecord({ id: "b1", tenantId: "b", agentId: "one" })); + + expect(send.mock.calls.map(([record]) => record.id)).toEqual(["a1", "b1"]); + expect(resolveToken.mock.calls.map(([record]) => record.id)).toEqual(["a1", "b1"]); + expect(persist.mock.calls.map(([record]) => record.id)).toEqual(["a1", "a2"]); + }); + + it("continues replay after a tenant's retryable failure and shares its backoff with live delivery", async () => { + const { claimBatch, complete, manager, persist, release, send } = createManager({ + options: { maxReplayBatchSize: 2 }, + }); + const first = makeClaim(makeRecord({ id: "a1", tenantId: "a" })); + const blocked = makeClaim(makeRecord({ id: "a2", tenantId: "a" })); + const healthy = makeClaim(makeRecord({ id: "b1", tenantId: "b" })); + claimBatch + .mockResolvedValueOnce([first]) + .mockResolvedValueOnce([blocked]) + .mockResolvedValueOnce([healthy]); + send.mockResolvedValueOnce({ + kind: "retryable", + correlationId: "throttled", + retryAfterMs: 60_000, + }); + + await manager.forceFlush(); + await manager.deliver(makeRecord({ id: "a3", tenantId: "a" })); + + expect(send.mock.calls.map(([record]) => record.id)).toEqual(["a1", "b1"]); + expect(complete.mock.calls.map(([claim]) => claim.record.id)).toEqual(["b1"]); + expect(release.mock.calls.map(([claim]) => claim.record.id)).toEqual(["a1", "a2"]); + expect(persist.mock.calls.map(([record]) => record.id)).toEqual(["a3"]); + }); + + it("scans past a backed-off prefix without spending the replay attempt budget", async () => { + const { claimBatch, complete, manager, release, send } = createManager({ + options: { maxReplayBatchSize: 1 }, + }); + send.mockResolvedValueOnce({ + kind: "retryable", + correlationId: "seed", + retryAfterMs: 60_000, + }); + await manager.deliver(makeRecord({ tenantId: "a" })); + for (let index = 0; index < 3; index++) { + claimBatch.mockResolvedValueOnce([ + makeClaim(makeRecord({ id: `blocked-${index}`, tenantId: "a" })), + ]); + } + claimBatch.mockResolvedValueOnce([makeClaim(makeRecord({ id: "healthy", tenantId: "b" }))]); + + await manager.forceFlush(); + + expect(complete.mock.calls.map(([claim]) => claim.record.id)).toEqual(["healthy"]); + expect(release.mock.calls).toHaveLength(3); + expect(send.mock.calls).toHaveLength(2); + }); + + it("bounds replay scans when every queued tenant is backed off", async () => { + const { claimBatch, manager, send } = createManager({ + options: { maxReplayBatchSize: 1 }, + }); + send.mockResolvedValueOnce({ + kind: "retryable", + correlationId: "seed", + retryAfterMs: 60_000, + }); + await manager.deliver(makeRecord()); + let claimed = 0; + claimBatch.mockImplementation(async () => [ + makeClaim(makeRecord({ id: `blocked-${claimed++}` })), + ]); + + await manager.forceFlush(); + + expect(claimBatch.mock.calls).toHaveLength(10); + expect(send.mock.calls).toHaveLength(1); + }); + it.each(["null", "throw", "timeout"] as const)( "persists live records when token resolution %s without backing off later sends", async (outcome) => { @@ -266,7 +355,9 @@ describe("DurableDeliveryManager", () => { }); const claim = makeClaim(makeRecord({ id: "released-replay-record" })); - claimBatch.mockResolvedValue([claim]); + claimBatch.mockImplementation(async (_limit, options) => + options?.excludeRecordIds?.has(claim.record.id) ? [] : [claim], + ); send.mockResolvedValue({ kind: "retryable", correlationId: "retryable-replay-record", @@ -275,72 +366,100 @@ describe("DurableDeliveryManager", () => { await manager.forceFlush(); - assert.deepEqual(claimBatch.mock.calls, [[1, { excludeRecordIds: new Set() }]]); + expect(claimBatch.mock.calls).toHaveLength(2); + expect(claimBatch.mock.calls[1][1]?.excludeRecordIds).toEqual(new Set([claim.record.id])); assert.deepEqual( release.mock.calls.map(([releasedClaim]) => releasedClaim), [claim], ); }); - it("excludes a released token-unavailable record from later claims in the same real-store pass while later records are still delivered", async () => { - const root = join( - process.cwd(), - ".superpowers", - "sdd", - "durable-manager-test-artifacts", - randomUUID(), - ); - await fs.mkdir(root, { recursive: true, mode: 0o700 }); - - try { - const logger = makeLogger(); - const store = await PersistentStore.create( - new ResolvedDurableDeliveryOptions({ enabled: true, storageDirectory: root }), - logger, + it.each(["token unavailable", "retryable response"] as const)( + "excludes a released %s record in a real-store pass while a later tenant is delivered", + async (failure) => { + const root = join( + process.cwd(), + ".superpowers", + "sdd", + "durable-manager-test-artifacts", + randomUUID(), ); + await fs.mkdir(root, { recursive: true, mode: 0o700 }); - const now = Date.now(); - const unavailable = makeRecord({ id: "token-unavailable-record", createdAt: now }); - const deliverable = makeRecord({ id: "deliverable-record", createdAt: now + 1 }); - await store.persist(unavailable); - await store.persist(deliverable); - - const tokenCallCounts = new Map(); - const resolveToken = vi.fn(async (record: DurableRecordV1) => { - tokenCallCounts.set(record.id, (tokenCallCounts.get(record.id) ?? 0) + 1); - return record.id === unavailable.id ? null : "fresh-token"; - }); - const send = vi.fn(async (record: DurableRecordV1) => ({ - kind: "success" as const, - correlationId: record.id, - })); - - const manager = new DurableDeliveryManager( - new ResolvedDurableDeliveryOptions({ - enabled: true, - maxReplayBatchSize: 5, - replayIntervalMilliseconds: 60_000, - }), - store, - logger, - { resolveToken, send }, - ); - managedInstances.push(manager); + try { + const logger = makeLogger(); + const store = await PersistentStore.create( + new ResolvedDurableDeliveryOptions({ enabled: true, storageDirectory: root }), + logger, + ); + + const now = Date.now(); + const unavailable = makeRecord({ id: "unavailable-record", createdAt: now, tenantId: "a" }); + const deliverable = makeRecord({ + id: "deliverable-record", + createdAt: now + 1, + tenantId: "b", + }); + await store.persist(unavailable); + await store.persist(deliverable); + + const tokenCallCounts = new Map(); + const resolveToken = vi.fn(async (record: DurableRecordV1) => { + tokenCallCounts.set(record.id, (tokenCallCounts.get(record.id) ?? 0) + 1); + return failure === "token unavailable" && record.id === unavailable.id + ? null + : "fresh-token"; + }); + const send = vi.fn(async (record: DurableRecordV1): Promise => + record.id === unavailable.id + ? { kind: "retryable", correlationId: record.id, retryAfterMs: 60_000 } + : { kind: "success", correlationId: record.id }, + ); + + const manager = new DurableDeliveryManager( + new ResolvedDurableDeliveryOptions({ + enabled: true, + maxReplayBatchSize: 5, + replayIntervalMilliseconds: 60_000, + }), + store, + logger, + { resolveToken, send }, + ); + managedInstances.push(manager); + + await manager.forceFlush(); + + assert.strictEqual(tokenCallCounts.get(unavailable.id), 1); + assert.deepEqual( + send.mock.calls.map(([record]) => record.id), + failure === "token unavailable" ? [deliverable.id] : [unavailable.id, deliverable.id], + ); + + const remaining = await store.claimBatch(5); + assert.strictEqual(remaining.length, 1); + assert.strictEqual(remaining[0].record.id, unavailable.id); + } finally { + await fs.rm(root, { recursive: true, force: true }); + } + }, + ); - await manager.forceFlush(); + it("stops replay after failing to release a token-unavailable record", async () => { + const { claimBatch, logger, manager, release, resolveToken, send } = createManager(); + claimBatch.mockResolvedValueOnce([makeClaim(makeRecord({ id: "first" }))]); + claimBatch.mockResolvedValueOnce([makeClaim(makeRecord({ id: "second" }))]); + resolveToken.mockResolvedValueOnce(null); + release.mockRejectedValueOnce(new Error("storage unavailable")); - assert.strictEqual(tokenCallCounts.get(unavailable.id), 1); - assert.deepEqual( - send.mock.calls.map(([record]) => record.id), - [deliverable.id], - ); + await manager.forceFlush(); - const remaining = await store.claimBatch(5); - assert.strictEqual(remaining.length, 1); - assert.strictEqual(remaining[0].record.id, unavailable.id); - } finally { - await fs.rm(root, { recursive: true, force: true }); - } + expect(claimBatch).toHaveBeenCalledTimes(1); + expect(send).not.toHaveBeenCalled(); + expect(logger.error).toHaveBeenCalledWith( + "[DurableDeliveryManager] Failed to release durable record", + expect.any(Error), + ); }); it("releases replay claims without sending while the transmission gate is backing off", async () => { @@ -855,7 +974,9 @@ function createManager( ) { const logger = makeLogger(); const persist = vi.fn(async (_record: DurableRecordV1) => "record.pending"); - const claimBatch = vi.fn(async (_limit: number) => [] as ClaimedRecord[]); + const claimBatch = vi.fn( + async (_limit: number, _options?: ClaimBatchOptions) => [] as ClaimedRecord[], + ); const complete = vi.fn(async (_claim: ClaimedRecord) => undefined); const release = vi.fn(async (_claim: ClaimedRecord) => undefined); const resolveToken = vi.fn(async (_record: DurableRecordV1) => "token"); diff --git a/test/internal/unit/a365/durableRecord.test.ts b/test/internal/unit/a365/durableRecord.test.ts index 0a99b454c..fbe72322e 100644 --- a/test/internal/unit/a365/durableRecord.test.ts +++ b/test/internal/unit/a365/durableRecord.test.ts @@ -71,6 +71,17 @@ describe("DurableRecord", () => { assert.throws(() => parseDurableRecord('{"version":2}'), /Unsupported durable record version/); }); + it.each(["", " \t"])("rejects blank tenant identifiers: %j", (tenantId) => { + const record = createDurableRecord({ + tenantId, + agentId: "agent", + useS2SEndpoint: false, + body: "{}", + }); + + assert.throws(() => parseDurableRecord(JSON.stringify(record)), /Invalid durable record/); + }); + it("rejects invalid record shapes", () => { assert.throws( () => diff --git a/test/internal/unit/a365/messageUtils.test.ts b/test/internal/unit/a365/messageUtils.test.ts index 6fb509110..74412af80 100644 --- a/test/internal/unit/a365/messageUtils.test.ts +++ b/test/internal/unit/a365/messageUtils.test.ts @@ -442,6 +442,22 @@ describe("serializeToolPayload", () => { }); }); + it("preserves explicitly empty metadata at the top level and in nested models", () => { + const payload = new ExecuteToolCallArguments({ + extension_data: {}, + resources: [{ extension_data: {}, container: { extension_data: {} } }], + }); + + expect(JSON.parse(serializeToolPayload(payload) as string)).toEqual({ + schema_version: "1.0", + metadata: {}, + resources: [{ metadata: {}, container: { metadata: {} } }], + }); + expect( + JSON.parse(serializeToolPayload(new ExecuteToolCallResult({ extension_data: {} })) as string), + ).toEqual({ schema_version: "1.0", metadata: {} }); + }); + it("omits explicit null schema versions", () => { const payload = new ExecuteToolCallArguments({ schema_version: null as any }); diff --git a/test/internal/unit/a365/persistentStore.test.ts b/test/internal/unit/a365/persistentStore.test.ts index dd1688458..23230014f 100644 --- a/test/internal/unit/a365/persistentStore.test.ts +++ b/test/internal/unit/a365/persistentStore.test.ts @@ -464,6 +464,44 @@ describe("PersistentStore", () => { assert.notStrictEqual(reclaimed.leasePath, claim.leasePath); }); + it("skips excluded records before leasing them while claiming a later record", async () => { + const root = join(scratchRoot, "excluded-records-store"); + const store = await createStore(root); + const now = Date.now(); + const excluded = makeRecord({ id: "excluded", createdAt: now }); + const deliverable = makeRecord({ id: "deliverable", createdAt: now + 1 }); + const excludedPath = await store.persist(excluded); + await store.persist(deliverable); + const rename = vi.spyOn(fs, "rename"); + + const [claim] = await store.claimBatch(1, { excludeRecordIds: new Set([excluded.id]) }); + + assert.strictEqual(claim.record.id, deliverable.id); + expect(rename).toHaveBeenCalledTimes(1); + assert.isTrue(await pathExists(excludedPath)); + }); + + it("still excludes content replaced between preflight and the atomic claim", async () => { + const root = join(scratchRoot, "exclusion-race-store"); + const store = await createStore(root); + const record = makeRecord({ id: "original" }); + const replaced = makeRecord({ id: "excluded" }); + const pendingPath = await store.persist(record); + const actualRename = fs.rename.bind(fs); + vi.spyOn(fs, "rename").mockImplementation(async (source, destination) => { + if (source === pendingPath) { + await fs.writeFile(pendingPath, JSON.stringify(replaced), "utf8"); + } + await actualRename(source, destination); + }); + + const claims = await store.claimBatch(1, { excludeRecordIds: new Set([replaced.id]) }); + + assert.deepEqual(claims, []); + const [retained] = await store.claimBatch(1); + assert.strictEqual(retained.record.id, replaced.id); + }); + it("recovers leases after their acquisition timestamp expires", async () => { let now = 1_000_000; vi.spyOn(Date, "now").mockImplementation(() => now); @@ -688,6 +726,31 @@ describe("PersistentStore", () => { assert.isTrue(files.some((file) => file.endsWith(".quarantine"))); }); + it.each(["", " \t"])( + "quarantines a blank tenant %j before claiming a later valid record", + async (tenantId) => { + const root = join(scratchRoot, "blank-tenant-store"); + const logger = makeLogger(); + const store = await createStore(root, {}, logger); + const valid = makeRecord(); + await store.persist(valid); + const badPath = join(explicitStorageRoot(root), "1-blank-tenant.pending"); + await fs.writeFile(badPath, JSON.stringify(makeRecord({ tenantId })), "utf8"); + + const [claim] = await store.claimBatch(1); + + assert.strictEqual(claim.record.id, valid.id); + assert.isFalse(await pathExists(badPath)); + assert.isTrue((await listStoreFiles(root)).some((file) => file.endsWith(".quarantine"))); + expect(logger.warn).toHaveBeenCalledWith( + "[PersistentStore] Quarantined malformed pending durable record", + expect.any(Error), + ); + await store.complete(claim); + assert.deepEqual(await store.claimBatch(1), []); + }, + ); + it.each(["../escape", "..\\escape"])( "quarantines records with unsafe ids (%j) to a short bounded name without creating files outside the storage root", async (unsafeId) => { diff --git a/test/internal/unit/a365/scopes.test.ts b/test/internal/unit/a365/scopes.test.ts index a68392b26..a68f0d0cd 100644 --- a/test/internal/unit/a365/scopes.test.ts +++ b/test/internal/unit/a365/scopes.test.ts @@ -22,6 +22,8 @@ import { OpenTelemetryConstants, ToolCallAction, ToolCallOutcomeStatus, + TransferMode, + TransferTargetType, } from "../../../../src/a365/index.js"; import type { AgentDetails, @@ -489,6 +491,126 @@ describe("Scopes", () => { }); describe("ExecuteToolScope", () => { + it("records the full tool URL without changing host and port attributes", () => { + const scope = ExecuteToolScope.start( + testRequest, + { + toolName: "search", + endpoint: { + host: "tools.example.com", + port: 8443, + url: "https://tools.example.com:8443/search?q=hello#results", + }, + }, + testAgentDetails, + ); + scope.dispose(); + + const attributes = sharedExporter.getFinishedSpans().at(-1)!.attributes; + expect(attributes["url.full"]).toBe("https://tools.example.com:8443/search?q=hello#results"); + expect(attributes["server.address"]).toBe("tools.example.com"); + expect(attributes["server.port"]).toBe(8443); + }); + + it("does not invent a URL for a host-only tool endpoint", () => { + const scope = ExecuteToolScope.start( + testRequest, + { toolName: "search", endpoint: { host: "tools.example.com", port: 443 } }, + testAgentDetails, + ); + scope.dispose(); + + const attributes = sharedExporter.getFinishedSpans().at(-1)!.attributes; + expect(attributes["url.full"]).toBeUndefined(); + expect(attributes["server.address"]).toBe("tools.example.com"); + expect(attributes["server.port"]).toBeUndefined(); + expect(attributes["microsoft.a365.transfer.mode"]).toBeUndefined(); + expect(attributes["microsoft.a365.transfer.target.type"]).toBeUndefined(); + }); + + it("records target-agent transfer attributes without replacing the source agent", () => { + const scope = ExecuteToolScope.start( + testRequest, + { + toolName: "handoff", + transferDetails: { + mode: TransferMode.PASS_CONTROL, + targetAgentDetails: { + agentId: "target-agent", + agentName: "Target Agent", + agentBlueprintId: "target-blueprint", + platformId: "target-platform", + agentVersion: "2.0", + tenantId: "target-tenant", + agentEmail: "not-emitted@example.com", + }, + }, + }, + testAgentDetails, + ); + scope.dispose(); + + const attributes = sharedExporter.getFinishedSpans().at(-1)!.attributes; + expect(attributes).toMatchObject({ + "gen_ai.agent.id": testAgentDetails.agentId, + "gen_ai.agent.name": testAgentDetails.agentName, + "microsoft.tenant.id": testAgentDetails.tenantId, + "microsoft.a365.transfer.mode": "pass_control", + "microsoft.a365.transfer.target.type": "agent", + "microsoft.a365.transfer.target.agent.id": "target-agent", + "microsoft.a365.transfer.target.agent.name": "Target Agent", + "microsoft.a365.transfer.target.agent.blueprint.id": "target-blueprint", + "microsoft.a365.transfer.target.agent.platform.id": "target-platform", + "microsoft.a365.transfer.target.agent.version": "2.0", + }); + expect(attributes["microsoft.a365.transfer.target.agent.email"]).toBeUndefined(); + expect(attributes["microsoft.a365.transfer.target.agent.tenant.id"]).toBeUndefined(); + }); + + it.each([TransferTargetType.AGENT, TransferTargetType.HUMAN, TransferTargetType.WORKFLOW])( + "records a %s transfer target without agent identity", + (targetType) => { + const scope = ExecuteToolScope.start( + testRequest, + { + toolName: "handoff", + transferDetails: { mode: TransferMode.RETURN_TO_CALLER, targetType }, + }, + testAgentDetails, + ); + scope.dispose(); + + const attributes = sharedExporter.getFinishedSpans().at(-1)!.attributes; + expect(attributes["microsoft.a365.transfer.mode"]).toBe("return_to_caller"); + expect(attributes["microsoft.a365.transfer.target.type"]).toBe(targetType); + expect(attributes["microsoft.a365.transfer.target.agent.id"]).toBeUndefined(); + }, + ); + + it.each([ + { mode: "invalid" }, + { mode: "pass_control", targetType: "invalid" }, + { + mode: "pass_control", + targetType: "human", + targetAgentDetails: { agentId: "target" }, + }, + ])("rejects invalid transfer details before starting a span: %j", (transferDetails) => { + const startSpan = vi.spyOn(trace.getTracer(OpenTelemetryConstants.SOURCE_NAME), "startSpan"); + try { + expect(() => + ExecuteToolScope.start( + testRequest, + { toolName: "handoff", transferDetails } as ToolCallDetails, + testAgentDetails, + ), + ).toThrow(/transfer/i); + expect(startSpan).not.toHaveBeenCalled(); + } finally { + startSpan.mockRestore(); + } + }); + it("should create scope with tool details", () => { const spy = vi.spyOn(OpenTelemetryScope.prototype as any, "setTagMaybe"); const callerDetails: UserDetails = { diff --git a/test/internal/unit/a365/transmissionGateRegistry.test.ts b/test/internal/unit/a365/transmissionGateRegistry.test.ts new file mode 100644 index 000000000..63c9e11cf --- /dev/null +++ b/test/internal/unit/a365/transmissionGateRegistry.test.ts @@ -0,0 +1,102 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +import { describe, expect, it } from "vitest"; +import { TransmissionGateRegistry } from "../../../../src/a365/exporter/durable/TransmissionGateRegistry.js"; + +describe("TransmissionGateRegistry", () => { + it("shares a tenant gate across concurrent leases but not across tenants", () => { + const registry = new TransmissionGateRegistry(); + const first = registry.acquire("a"); + const second = registry.acquire("a"); + const other = registry.acquire("b"); + first.gate.recordRetryableFailure(first.gate.acquire()!, 60_000); + + expect(second.gate.canAcquire()).toBe(false); + expect(other.gate.canAcquire()).toBe(true); + first.release(); + second.release(); + other.release(); + }); + + it("does not evict an active tenant under capacity pressure", () => { + const registry = new TransmissionGateRegistry({ capacity: 1 }); + const first = registry.acquire("a"); + const second = registry.acquire("b"); + const again = registry.acquire("a"); + + expect(again.gate).toBe(first.gate); + again.release(); + second.release(); + const replacement = registry.acquire("b"); + expect(replacement.gate).not.toBe(second.gate); + replacement.release(); + first.release(); + }); + + it("preserves pending backoff under capacity and idle pressure", () => { + let now = 0; + const registry = new TransmissionGateRegistry({ + capacity: 1, + idleTimeoutMilliseconds: 10, + now: () => now, + }); + const blocked = registry.acquire("a"); + blocked.gate.recordRetryableFailure(blocked.gate.acquire()!, 60_000); + blocked.release(); + now = 20; + const other = registry.acquire("b"); + other.release(); + const again = registry.acquire("a"); + + expect(again.gate).toBe(blocked.gate); + expect(again.gate.canAcquire()).toBe(false); + again.release(); + + now = 60_000; + const expired = registry.acquire("a"); + expect(expired.gate).not.toBe(blocked.gate); + expired.release(); + }); + + it("evicts least recently used inactive gates and tolerates repeated release", () => { + let now = 0; + const registry = new TransmissionGateRegistry({ capacity: 2, now: () => now }); + const oldest = registry.acquire("a"); + oldest.release(); + now++; + const recent = registry.acquire("b"); + recent.release(); + now++; + const third = registry.acquire("c"); + third.release(); + oldest.release(); + const retained = registry.acquire("b"); + expect(retained.gate).toBe(recent.gate); + retained.release(); + const evicted = registry.acquire("a"); + expect(evicted.gate).not.toBe(oldest.gate); + evicted.release(); + }); + + it("removes idle gates without waiting for capacity pressure", () => { + let now = 0; + const registry = new TransmissionGateRegistry({ + idleTimeoutMilliseconds: 10, + now: () => now, + }); + const first = registry.acquire("a"); + first.release(); + now = 10; + const again = registry.acquire("a"); + expect(again.gate).not.toBe(first.gate); + again.release(); + }); + + it("rejects invalid limits and empty tenant identifiers", () => { + expect(() => new TransmissionGateRegistry({ capacity: 0 })).toThrow(RangeError); + expect(() => new TransmissionGateRegistry({ capacity: 1.5 })).toThrow(RangeError); + expect(() => new TransmissionGateRegistry({ idleTimeoutMilliseconds: 0 })).toThrow(RangeError); + expect(() => new TransmissionGateRegistry().acquire(" ")).toThrow(TypeError); + }); +}); From b981155cf0fd03042955361acffd37a7057fc398 Mon Sep 17 00:00:00 2001 From: "Nikhil Chitlur Navakiran (from Dev Box)" Date: Fri, 30 Oct 2026 10:43:24 -0600 Subject: [PATCH 2/2] fix(a365): skip backed-off tenants before replay claims Prevent blocked prefixes larger than the per-pass scan budget from starving later tenants. Preserve post-claim exclusion checks and cover repeated throttling passes, tenant recovery, and claim-time replacement races. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 19b76a8f-e714-42c6-ac3f-a2905b3f2411 --- CHANGELOG.md | 2 +- .../durable/DurableDeliveryManager.ts | 5 +- src/a365/exporter/durable/PersistentStore.ts | 10 ++- .../durable/TransmissionGateRegistry.ts | 10 +++ .../unit/a365/durableDeliveryManager.test.ts | 83 +++++++++++++++++- .../unit/a365/persistentStore.test.ts | 84 +++++++++++-------- .../a365/transmissionGateRegistry.test.ts | 18 ++++ 7 files changed, 170 insertions(+), 42 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4681a2791..99fb1f55e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,7 @@ ### Bugs Fixed - Isolate Agent365 durable delivery backoff by tenant, shared across that tenant's agents and live/replay paths. Continue replay past retryable failures and backed-off tenants without spending the attempt limit on deferred records; bound replay scans and preserve active backoff during registry eviction. -- Skip excluded durable records before leasing and quarantine blank-tenant records before replay, retaining post-claim content validation. +- Skip excluded durable records and backed-off tenants before leasing so blocked prefixes larger than the scan budget cannot starve healthy tenants across replay passes. Quarantine blank-tenant records before replay, retaining post-claim content validation. - Preserve explicitly empty `extension_data` as nested `metadata: {}` in typed ExecuteTool payloads, including nested models, while continuing to omit absent/null metadata. - Update OpenAI Agents to 0.15.0, Vitest and its coverage provider to 4.1.12, MSAL Node to 7.1.1, and vulnerable transitive dependencies across the distro and samples. OpenAI Agents 0.15.0 removes the vulnerable MCP/Express dependency chain. - Require Node.js 22.23.4+, 24.22.1+, or 26.12.1+ on their respective release lines for security fixes. diff --git a/src/a365/exporter/durable/DurableDeliveryManager.ts b/src/a365/exporter/durable/DurableDeliveryManager.ts index 3df08042c..5d7fa8eaa 100644 --- a/src/a365/exporter/durable/DurableDeliveryManager.ts +++ b/src/a365/exporter/durable/DurableDeliveryManager.ts @@ -201,7 +201,10 @@ export class DurableDeliveryManager { let claim: ClaimedRecord | undefined; try { - [claim] = await this.store.claimBatch(1, { excludeRecordIds: skippedRecordIds }); + [claim] = await this.store.claimBatch(1, { + excludeRecordIds: skippedRecordIds, + excludeTenantIds: this.gates.getDeferredTenantIds(), + }); } catch (error) { this.logger.error("[DurableDeliveryManager] Failed to claim durable records", error); return; diff --git a/src/a365/exporter/durable/PersistentStore.ts b/src/a365/exporter/durable/PersistentStore.ts index b729ebb1a..593432dd5 100644 --- a/src/a365/exporter/durable/PersistentStore.ts +++ b/src/a365/exporter/durable/PersistentStore.ts @@ -40,6 +40,8 @@ export interface ClaimBatchOptions { * temporarily unusable without starving unrelated records that sort later. */ excludeRecordIds?: ReadonlySet; + /** Skip tenants whose transmission gates currently defer replay. */ + excludeTenantIds?: ReadonlySet; } interface ManagedFile { @@ -120,6 +122,7 @@ export class PersistentStore { await this.recoverStaleLeases(); const excludeRecordIds = options?.excludeRecordIds; + const excludeTenantIds = options?.excludeTenantIds; const claims: ClaimedRecord[] = []; const pendingFiles = await this.listManagedFilePaths((name) => name.endsWith(PENDING_SUFFIX)); pendingFiles.sort(); @@ -163,7 +166,10 @@ export class PersistentStore { try { const preflightRecord = parseDurableRecord(preflightText); - if (excludeRecordIds?.has(preflightRecord.id)) { + if ( + excludeRecordIds?.has(preflightRecord.id) || + excludeTenantIds?.has(preflightRecord.tenantId) + ) { continue; } } catch (error) { @@ -218,7 +224,7 @@ export class PersistentStore { continue; } - if (excludeRecordIds?.has(record.id)) { + if (excludeRecordIds?.has(record.id) || excludeTenantIds?.has(record.tenantId)) { // Leave this record for a later pass instead of starving records // that sort after it: release it back to pending and keep scanning. await this.release({ record, leasePath }); diff --git a/src/a365/exporter/durable/TransmissionGateRegistry.ts b/src/a365/exporter/durable/TransmissionGateRegistry.ts index 7a8c279b5..f3d2aa7ed 100644 --- a/src/a365/exporter/durable/TransmissionGateRegistry.ts +++ b/src/a365/exporter/durable/TransmissionGateRegistry.ts @@ -68,6 +68,16 @@ export class TransmissionGateRegistry { }; } + getDeferredTenantIds(): ReadonlySet { + const deferred = new Set(); + for (const [tenantId, entry] of this.entries) { + if (!entry.gate.canAcquire()) { + deferred.add(tenantId); + } + } + return deferred; + } + private trim(now: number): void { const inactive: [string, GateEntry][] = []; for (const [tenantId, entry] of this.entries) { diff --git a/test/internal/unit/a365/durableDeliveryManager.test.ts b/test/internal/unit/a365/durableDeliveryManager.test.ts index 95405927a..8e14c0e48 100644 --- a/test/internal/unit/a365/durableDeliveryManager.test.ts +++ b/test/internal/unit/a365/durableDeliveryManager.test.ts @@ -335,7 +335,9 @@ describe("DurableDeliveryManager", () => { await firstSendStarted.promise; try { - assert.deepEqual(claimBatch.mock.calls, [[1, { excludeRecordIds: new Set() }]]); + assert.deepEqual(claimBatch.mock.calls, [ + [1, { excludeRecordIds: new Set(), excludeTenantIds: new Set() }], + ]); assert.deepEqual(pendingClaims, [second]); } finally { finishFirstSend.resolve({ kind: "success", correlationId: first.record.id }); @@ -343,8 +345,8 @@ describe("DurableDeliveryManager", () => { } assert.deepEqual(claimBatch.mock.calls, [ - [1, { excludeRecordIds: new Set() }], - [1, { excludeRecordIds: new Set() }], + [1, { excludeRecordIds: new Set(), excludeTenantIds: new Set() }], + [1, { excludeRecordIds: new Set(), excludeTenantIds: new Set() }], ]); assert.deepEqual(pendingClaims, []); }); @@ -445,6 +447,77 @@ describe("DurableDeliveryManager", () => { }, ); + it("delivers later tenants across passes with a throttled prefix larger than the scan budget", async () => { + const root = join( + process.cwd(), + ".superpowers", + "sdd", + "durable-manager-test-artifacts", + randomUUID(), + ); + await fs.mkdir(root, { recursive: true, mode: 0o700 }); + let now = Date.now(); + vi.spyOn(Date, "now").mockImplementation(() => now); + + try { + const logger = makeLogger(); + const options = new ResolvedDurableDeliveryOptions({ + enabled: true, + storageDirectory: root, + maxReplayBatchSize: 2, + replayIntervalMilliseconds: 60_000, + }); + const store = await PersistentStore.create(options, logger); + for (let index = 0; index < 25; index++) { + await store.persist( + makeRecord({ id: `blocked-${index}`, createdAt: now + index, tenantId: "a" }), + ); + } + + let throttled = true; + const send = vi.fn(async (record: DurableRecordV1): Promise => + record.tenantId === "a" && throttled + ? { kind: "retryable", correlationId: record.id, status: 429, retryAfterMs: 60_000 } + : { kind: "success", correlationId: record.id }, + ); + const manager = new DurableDeliveryManager(options, store, logger, { + resolveToken: async () => "fresh-token", + send, + }); + managedInstances.push(manager); + const claimBatch = vi.spyOn(store, "claimBatch"); + const complete = vi.spyOn(store, "complete"); + + for (let pass = 0; pass < 3; pass++) { + await store.persist( + makeRecord({ id: `healthy-${pass}`, createdAt: now + 100, tenantId: "b" }), + ); + claimBatch.mockClear(); + await manager.forceFlush(); + expect(claimBatch.mock.calls.length).toBeLessThanOrEqual(20); + now += 60_001; + } + + expect(complete.mock.calls.map(([claim]) => claim.record.id)).toEqual([ + "healthy-0", + "healthy-1", + "healthy-2", + ]); + expect(send.mock.calls.filter(([record]) => record.tenantId === "a")).toHaveLength(3); + + throttled = false; + for (let pass = 0; pass < 13; pass++) { + await manager.forceFlush(); + } + expect(complete.mock.calls.filter(([claim]) => claim.record.tenantId === "a")).toHaveLength( + 25, + ); + expect(await store.claimBatch(1)).toEqual([]); + } finally { + await fs.rm(root, { recursive: true, force: true }); + } + }); + it("stops replay after failing to release a token-unavailable record", async () => { const { claimBatch, logger, manager, release, resolveToken, send } = createManager(); claimBatch.mockResolvedValueOnce([makeClaim(makeRecord({ id: "first" }))]); @@ -626,7 +699,9 @@ describe("DurableDeliveryManager", () => { release.mock.calls.map(([releasedClaim]) => releasedClaim), [invalid], ); - assert.deepEqual(claimBatch.mock.calls, [[1, { excludeRecordIds: new Set() }]]); + assert.deepEqual(claimBatch.mock.calls, [ + [1, { excludeRecordIds: new Set(), excludeTenantIds: new Set() }], + ]); }); it("closes the transmission gate after a permanent replay probe response", async () => { diff --git a/test/internal/unit/a365/persistentStore.test.ts b/test/internal/unit/a365/persistentStore.test.ts index 23230014f..80c40dd68 100644 --- a/test/internal/unit/a365/persistentStore.test.ts +++ b/test/internal/unit/a365/persistentStore.test.ts @@ -464,43 +464,59 @@ describe("PersistentStore", () => { assert.notStrictEqual(reclaimed.leasePath, claim.leasePath); }); - it("skips excluded records before leasing them while claiming a later record", async () => { - const root = join(scratchRoot, "excluded-records-store"); - const store = await createStore(root); - const now = Date.now(); - const excluded = makeRecord({ id: "excluded", createdAt: now }); - const deliverable = makeRecord({ id: "deliverable", createdAt: now + 1 }); - const excludedPath = await store.persist(excluded); - await store.persist(deliverable); - const rename = vi.spyOn(fs, "rename"); - - const [claim] = await store.claimBatch(1, { excludeRecordIds: new Set([excluded.id]) }); - - assert.strictEqual(claim.record.id, deliverable.id); - expect(rename).toHaveBeenCalledTimes(1); - assert.isTrue(await pathExists(excludedPath)); - }); + it.each(["record", "tenant"] as const)( + "skips an excluded %s before leasing while claiming a later record", + async (exclusion) => { + const root = join(scratchRoot, "excluded-records-store"); + const store = await createStore(root); + const now = Date.now(); + const excluded = makeRecord({ id: "excluded", createdAt: now, tenantId: "a" }); + const deliverable = makeRecord({ id: "deliverable", createdAt: now + 1, tenantId: "b" }); + const excludedPath = await store.persist(excluded); + await store.persist(deliverable); + const rename = vi.spyOn(fs, "rename"); + + const [claim] = await store.claimBatch( + 1, + exclusion === "record" + ? { excludeRecordIds: new Set([excluded.id]) } + : { excludeTenantIds: new Set([excluded.tenantId]) }, + ); - it("still excludes content replaced between preflight and the atomic claim", async () => { - const root = join(scratchRoot, "exclusion-race-store"); - const store = await createStore(root); - const record = makeRecord({ id: "original" }); - const replaced = makeRecord({ id: "excluded" }); - const pendingPath = await store.persist(record); - const actualRename = fs.rename.bind(fs); - vi.spyOn(fs, "rename").mockImplementation(async (source, destination) => { - if (source === pendingPath) { - await fs.writeFile(pendingPath, JSON.stringify(replaced), "utf8"); - } - await actualRename(source, destination); - }); + assert.strictEqual(claim.record.id, deliverable.id); + expect(rename).toHaveBeenCalledTimes(1); + assert.isTrue(await pathExists(excludedPath)); + }, + ); - const claims = await store.claimBatch(1, { excludeRecordIds: new Set([replaced.id]) }); + it.each(["record", "tenant"] as const)( + "still excludes %s content replaced between preflight and the atomic claim", + async (exclusion) => { + const root = join(scratchRoot, "exclusion-race-store"); + const store = await createStore(root); + const record = makeRecord({ id: "original", tenantId: "a" }); + const replaced = makeRecord({ id: "excluded", tenantId: "b" }); + const pendingPath = await store.persist(record); + const actualRename = fs.rename.bind(fs); + vi.spyOn(fs, "rename").mockImplementation(async (source, destination) => { + if (source === pendingPath) { + await fs.writeFile(pendingPath, JSON.stringify(replaced), "utf8"); + } + await actualRename(source, destination); + }); + + const claims = await store.claimBatch( + 1, + exclusion === "record" + ? { excludeRecordIds: new Set([replaced.id]) } + : { excludeTenantIds: new Set([replaced.tenantId]) }, + ); - assert.deepEqual(claims, []); - const [retained] = await store.claimBatch(1); - assert.strictEqual(retained.record.id, replaced.id); - }); + assert.deepEqual(claims, []); + const [retained] = await store.claimBatch(1); + assert.strictEqual(retained.record.id, replaced.id); + }, + ); it("recovers leases after their acquisition timestamp expires", async () => { let now = 1_000_000; diff --git a/test/internal/unit/a365/transmissionGateRegistry.test.ts b/test/internal/unit/a365/transmissionGateRegistry.test.ts index 63c9e11cf..48170622c 100644 --- a/test/internal/unit/a365/transmissionGateRegistry.test.ts +++ b/test/internal/unit/a365/transmissionGateRegistry.test.ts @@ -5,6 +5,24 @@ import { describe, expect, it } from "vitest"; import { TransmissionGateRegistry } from "../../../../src/a365/exporter/durable/TransmissionGateRegistry.js"; describe("TransmissionGateRegistry", () => { + it("reports deferred tenants until backoff expires or their half-open probe finishes", () => { + let now = 0; + const registry = new TransmissionGateRegistry({ now: () => now }); + const blocked = registry.acquire("a"); + const healthy = registry.acquire("b"); + blocked.gate.recordRetryableFailure(blocked.gate.acquire()!, 60_000); + + expect(registry.getDeferredTenantIds()).toEqual(new Set(["a"])); + now = 60_000; + expect(registry.getDeferredTenantIds()).toEqual(new Set()); + const probe = blocked.gate.acquire()!; + expect(registry.getDeferredTenantIds()).toEqual(new Set(["a"])); + blocked.gate.recordSuccess(probe); + expect(registry.getDeferredTenantIds()).toEqual(new Set()); + blocked.release(); + healthy.release(); + }); + it("shares a tenant gate across concurrent leases but not across tenants", () => { const registry = new TransmissionGateRegistry(); const first = registry.acquire("a");