Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion setup/js/actions_secret_masking.cjs
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ function maskSecret(value) {
if (!secret) return;

const setSecret = global.core?.setSecret;
if (typeof setSecret === "function" && !setSecret.__ghAwUnavailable) {
if (typeof setSecret === "function" && !("__ghAwUnavailable" in setSecret && setSecret.__ghAwUnavailable)) {
setSecret.call(global.core, secret);
return;
}
Expand Down
1 change: 1 addition & 0 deletions setup/js/add_workflow_run_comment.cjs
Original file line number Diff line number Diff line change
Expand Up @@ -233,6 +233,7 @@ async function updateReusableStatusComment(reusableComment, invocationContext, r
* Add a comment with a workflow run link to the triggering item.
* This script ONLY creates comments - it does NOT add reactions.
* Use add_reaction.cjs in the pre-activation job to add reactions first for immediate feedback.
* @param {Omit<typeof context, "issue" | "repo"> & {repo?: typeof context.repo, workflowEmoji?: string, nonFatalStatusCommentErrors?: boolean}} [rawContext]
*/
async function createOrReuseStatusComment(rawContext = context) {
const messagesConfig = getMessages();
Expand Down
65 changes: 52 additions & 13 deletions setup/js/agent_session.cjs
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ const { getMessageRefusal, getProviderRefusals, normalizeOpenAIChatUsage } = req
const USAGE_ALIASES = {
input_tokens: "inputTokens",
output_tokens: "outputTokens",
reasoning_output_tokens: "reasoningOutputTokens",
total_tokens: "totalTokens",
cache_creation_input_tokens: "cacheCreationInputTokens",
cache_read_input_tokens: "cacheReadInputTokens",
Expand All @@ -32,6 +33,35 @@ function isMetric(value) {
return typeof value === "number" && Number.isFinite(value) && value >= 0;
}

/**
* Subagent lifecycle envelope IDs identify the target, not the emitting caller.
* @param {any} event
* @returns {{sessionId?: string|null, parentToolUseId?: string|null, agentId?: string}}
*/
function sessionContext(event) {
const data = event?.data ?? {};
const sessionId = data.sessionId !== undefined ? data.sessionId : event?.sessionId !== undefined ? event.sessionId : event?.session_id !== undefined ? event.session_id : data.session_id;
const parentToolUseId =
data.parentToolUseId !== undefined ? data.parentToolUseId : event?.parentToolUseId !== undefined ? event.parentToolUseId : event?.parent_tool_use_id !== undefined ? event.parent_tool_use_id : data.parent_tool_use_id;
const agentId = data.agentId ?? (typeof event?.type === "string" && event.type.startsWith("subagent.") ? undefined : event?.agentId);
return {
...(typeof sessionId === "string" || sessionId === null ? { sessionId } : {}),
...(typeof parentToolUseId === "string" || parentToolUseId === null ? { parentToolUseId } : {}),
...(typeof agentId === "string" ? { agentId } : {}),
};
}

/** @param {Array<any>} events @returns {boolean} */
function isSingleNestedSession(events) {
return events.length > 0 && events.every(event => !!sessionContext(event).parentToolUseId) && new Set(events.map(sessionScopeKey)).size === 1;
}

/** @param {any} event @returns {string} */
function sessionScopeKey(event) {
const { sessionId, parentToolUseId, agentId } = sessionContext(event);
return JSON.stringify([sessionId, parentToolUseId, agentId]);
}

/** @param {any} usage @returns {Record<string, any>|undefined} */
function normalizeSessionUsage(usage) {
// Usage is a named-field object; arrays cannot represent its token fields.
Expand Down Expand Up @@ -331,19 +361,22 @@ function normalizeAgentSession(entries, { sourceEngine } = {}) {
/**
* Select snapshots field by field, never sum snapshots or erase prior known values.
* @param {Array<any>} events
* @param {{includeNested?: boolean}} [options]
* @returns {Record<string, any>|undefined}
*/
function selectSessionResult(events) {
function selectSessionResult(events, { includeNested = false } = {}) {
const normalized = normalizeAgentSession(events);
const claudeSessions = new Map();
for (const event of normalized) {
if (event.type !== "session.result" || event.data.sourceEngine !== "claude" || event.parent_tool_use_id || typeof event.session_id !== "string") continue;
claudeSessions.set(event.session_id, [...(claudeSessions.get(event.session_id) ?? []), { ...event, session_id: undefined }]);
const context = sessionContext(event);
if (event.type !== "session.result" || event.data.sourceEngine !== "claude" || (!includeNested && context.parentToolUseId) || typeof context.sessionId !== "string") continue;
const key = sessionScopeKey(event);
claudeSessions.set(key, [...(claudeSessions.get(key) ?? []), event]);
}
if (claudeSessions.size > 1) {
const aggregate = { usage: {} };
for (const observations of claudeSessions.values()) {
const snapshot = selectSessionResult(observations);
const snapshot = selectSessionResult(observations, { includeNested });
if (!snapshot) continue;
accumulateSessionUsage(aggregate.usage, snapshot.usage);
for (const key of ["numTurns", "durationMs", "totalCostUsd"]) {
Expand All @@ -354,7 +387,8 @@ function selectSessionResult(events) {
}
}
for (const event of normalized) {
if (event.type !== "session.result" || event.data.sourceEngine !== "claude" || event.parent_tool_use_id || typeof event.session_id === "string") continue;
const context = sessionContext(event);
if (event.type !== "session.result" || event.data.sourceEngine !== "claude" || (!includeNested && context.parentToolUseId) || typeof context.sessionId === "string") continue;
// Unassigned diagnostics are valid evidence, but their usage may overlap a named session.
for (const key of ["errors", "permissionDenials"]) {
if (Array.isArray(event.data[key])) aggregate[key] = [...(aggregate[key] ?? []), ...structuredClone(event.data[key])];
Expand All @@ -367,7 +401,7 @@ function selectSessionResult(events) {
let result;
for (const event of normalized) {
if (event.type !== "session.result") continue;
if (event.data.sourceEngine === "claude" && event.parent_tool_use_id) continue;
if (!includeNested && event.data.sourceEngine === "claude" && sessionContext(event).parentToolUseId) continue;
result ??= {};
const data = event.data;
for (const [key, value] of Object.entries(data)) {
Expand All @@ -384,9 +418,9 @@ function selectSessionResult(events) {
return result;
}

/** @param {Array<any>} events @returns {any|undefined} */
function projectSessionResult(events) {
const data = selectSessionResult(events);
/** @param {Array<any>} events @param {{includeNested?: boolean}} [options] @returns {any|undefined} */
function projectSessionResult(events, options) {
const data = selectSessionResult(events, options);
if (!data) return undefined;
return {
type: "result",
Expand All @@ -399,11 +433,12 @@ function projectSessionResult(events) {
};
}

/** @param {Array<any>} events @returns {any|undefined} */
function projectSessionInitialization(events) {
/** @param {Array<any>} events @param {{includeNested?: boolean}} [options] @returns {any|undefined} */
function projectSessionInitialization(events, { includeNested = false } = {}) {
let data;
for (const event of normalizeAgentSession(events)) {
if (event.type !== "session.init" && event.type !== "session.start") continue;
if (!includeNested && sessionContext(event).parentToolUseId) continue;
data ??= {};
for (const [key, value] of Object.entries(event.data)) {
if (value !== undefined) data = { ...data, [key]: structuredClone(value) };
Expand Down Expand Up @@ -531,17 +566,21 @@ function sessionTokenTotal(usage) {
return isTokenCount(total) ? total : undefined;
}

/** @param {Array<any>} events @returns {string|undefined} */
function observedSessionModel(events) {
/** @param {Array<any>} events @param {{includeNested?: boolean}} [options] @returns {string|undefined} */
function observedSessionModel(events, { includeNested = false } = {}) {
let model;
for (const event of events) {
if (!includeNested && sessionContext(event).parentToolUseId) continue;
const value = event?.type === "system" && event.subtype === "init" ? event.model : ["session.init", "session.start", "pi.message_snapshot"].includes(event?.type) ? event.data?.model : undefined;
if (typeof value === "string" && value.length) model = value;
}
return model;
}

module.exports = {
sessionContext,
sessionScopeKey,
isSingleNestedSession,
isSessionEvent,
isTokenCount,
isMetric,
Expand Down
2 changes: 1 addition & 1 deletion setup/js/agent_session_render.cjs
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,7 @@ function collapseStreamedMessages(events) {
const key = JSON.stringify([scope, type, messageId ?? claude?.id, claude?.sequence, position]);
const previousText = previous?.legacy ? previous.entry.message.content[0][previous.field] : previous?.entry.data.content;
const geminiContinuation = previous?.key === key && geminiSnapshot?.scope === scope && geminiSnapshot.id === messageId && redactManifestValue(previousText + text, secrets) === redactManifestValue(geminiSnapshot.content, secrets);
const continuation = (native && ["content_block_start", "content_block_delta"].includes(native.type)) || claudeContinuation || geminiContinuation;
const continuation = source.data?.delta !== false && ((native && ["content_block_start", "content_block_delta"].includes(native.type)) || claudeContinuation || geminiContinuation);
const streamed = delta || continuation || (partial && messageId !== undefined);
geminiSnapshot = undefined;
if (!["assistant.message", "assistant.reasoning"].includes(type) || typeof text !== "string" || (!streamed && !(messageId !== undefined && previous?.key === key))) {
Expand Down
113 changes: 71 additions & 42 deletions setup/js/aw_context.cjs
Original file line number Diff line number Diff line change
Expand Up @@ -167,58 +167,76 @@ function readInboundAwContext(payload) {
return parseInboundAwContext(payload?.inputs?.aw_context) || parseInboundAwContext(payload?.client_payload?.aw_context);
}

/**
* Resolve the compiler-managed workflow_dispatch work queue claim, falling
* back to the legacy assignment embedded in aw_context.
*
* @param {any} payload
* @returns {{work_id: string, claim_id: string, work: Record<string, unknown>} | null}
*/
function readInboundWorkQueueAssignment(payload) {
const rawClaim = payload?.inputs?.work_queue_claim ?? payload?.client_payload?.work_queue_claim;
const contextAssignment = readWorkQueueAssignment(readInboundAwContext(payload));
if (rawClaim == null || rawClaim === "") return contextAssignment;
function parseWorkQueueAssignment(raw) {
const { parseStrictJSON } = require("./work_queue_codec.cjs");
const { normalizeAssignment } = require("./work_queue_claim_scope.cjs");
return normalizeAssignment(typeof raw === "string" ? parseStrictJSON(raw) : raw);
}

const claim = parseInboundAwContext(rawClaim);
if (!claim) throw new TypeError("work_queue_claim must be a JSON object");
const inputAssignment = readWorkQueueAssignment({ work_queue: claim });
if (contextAssignment) {
throw new TypeError("work_queue_claim cannot be combined with an assignment in aw_context");
function readInboundWorkQueueAssignment(payload) {
const { parseStrictJSON } = require("./work_queue_codec.cjs");
const assignments = [];
for (const source of [payload?.inputs, payload?.client_payload]) {
if (!source) continue;
if (!isRecord(source)) throw new TypeError("work queue input context must be an object");
if (["work_queue_claim", "work_claim", "work_queue"].some(key => Object.hasOwn(source, key))) throw new TypeError("legacy scalar work queue assignments are unsupported");
if (Object.hasOwn(source, "work_queue_assignment")) assignments.push(parseWorkQueueAssignment(source.work_queue_assignment));
if (source.aw_context !== undefined && source.aw_context !== null && source.aw_context !== "") {
const embeddedContext = typeof source.aw_context === "string" ? parseStrictJSON(source.aw_context) : source.aw_context;
if (!isRecord(embeddedContext)) throw new TypeError("work queue input context must be an object");
const embedded = readWorkQueueAssignment(embeddedContext);
if (embedded) assignments.push(embedded);
}
}
return inputAssignment;
if (assignments.length > 1) throw new TypeError("work_queue_assignment must have exactly one source");
return assignments[0] || null;
}

function resolveWorkQueueRuntime(payload, { role = process.env.GH_AW_WORK_QUEUE_ROLE, requireAssignment = false } = {}) {
const assignment = readInboundWorkQueueAssignment(payload);
if (role !== undefined && !["observer", "dispatcher", "worker"].includes(role)) throw new TypeError("work_queue_runtime_role_invalid");
if (requireAssignment && role !== undefined && role !== "worker") throw new Error("work_queue_runtime_role_conflict");
if (role === "observer" && assignment) throw new Error("work_queue_observer_assignment_forbidden");
if (role === "dispatcher" && assignment) throw new Error("work_queue_runtime_role_conflict");
const resolvedRole = role ?? (assignment || requireAssignment ? "worker" : "dispatcher");
if (resolvedRole === "worker" && !assignment) throw new Error("work_queue_assignment_required");
return { role: resolvedRole, assignment };
}

/**
* @param {Record<string, unknown> | null} awContext
* @returns {{work_id: string, claim_id: string, work: Record<string, unknown>} | null}
*/
function readWorkQueueAssignment(awContext) {
if (!awContext) return null;
const hasCurrent = Object.hasOwn(awContext, "work_queue");
const hasLegacy = Object.hasOwn(awContext, "work_claim");
if (hasCurrent && hasLegacy) {
throw new TypeError("aw_context cannot contain both work_queue and the legacy work_claim assignment");
}
if (!hasCurrent && !hasLegacy) return null;
const assignment = hasCurrent ? awContext.work_queue : awContext.work_claim;
if (!isRecord(assignment) || Object.keys(assignment).length !== 3 || typeof assignment.work_id !== "string" || !assignment.work_id || typeof assignment.claim_id !== "string" || !assignment.claim_id || !isRecord(assignment.work)) {
throw new TypeError("work queue assignment has an invalid shape");
}
return { work_id: assignment.work_id, claim_id: assignment.claim_id, work: assignment.work };
if (["work_queue", "work_claim", "work_queue_claim"].some(key => Object.hasOwn(awContext, key))) throw new TypeError("legacy scalar work queue assignments are unsupported");
return Object.hasOwn(awContext, "work_queue_assignment") ? parseWorkQueueAssignment(awContext.work_queue_assignment) : null;
}

/**
* @param {Record<string, unknown>} awContext
* @param {unknown} [rawWorkQueueClaim]
* @returns {Record<string, unknown>}
*/
function normalizeWorkQueueContext(awContext, rawWorkQueueClaim) {
function normalizeWorkQueueContext(awContext, rawWorkQueueAssignment) {
let assignment = readWorkQueueAssignment(awContext);
if (rawWorkQueueClaim != null && rawWorkQueueClaim !== "") {
const claim = parseInboundAwContext(rawWorkQueueClaim);
if (rawWorkQueueAssignment !== undefined) {
if (assignment) throw new TypeError("work_queue_assignment must have exactly one source");
assignment = parseWorkQueueAssignment(rawWorkQueueAssignment);
}
return assignment ? { ...awContext, work_queue_assignment: assignment } : { ...awContext };
}

// Preserve pre-queue metadata normalization only for workflows without the
// compiler-managed queue. This data never supplies current Claim authority.
function normalizeLegacyWorkQueueContext(awContext, rawClaim) {
const readAssignment = value => {
const hasCurrent = Object.hasOwn(value, "work_queue");
const hasLegacy = Object.hasOwn(value, "work_claim");
if (hasCurrent && hasLegacy) throw new TypeError("aw_context cannot contain both work_queue and the legacy work_claim assignment");
if (!hasCurrent && !hasLegacy) return null;
const assignment = hasCurrent ? value.work_queue : value.work_claim;
if (!isRecord(assignment) || Object.keys(assignment).length !== 3 || typeof assignment.work_id !== "string" || !assignment.work_id || typeof assignment.claim_id !== "string" || !assignment.claim_id || !isRecord(assignment.work))
throw new TypeError("work queue assignment has an invalid shape");
return { work_id: assignment.work_id, claim_id: assignment.claim_id, work: assignment.work };
};
let assignment = readAssignment(awContext);
if (rawClaim != null && rawClaim !== "") {
const claim = parseInboundAwContext(rawClaim);
if (!claim) throw new TypeError("work_queue_claim must be a JSON object");
if (assignment) throw new TypeError("work_queue_claim cannot be combined with an assignment in aw_context");
assignment = readWorkQueueAssignment({ work_queue: claim });
assignment = readAssignment({ work_queue: claim });
}
const normalized = { ...awContext };
delete normalized.work_claim;
Expand Down Expand Up @@ -416,4 +434,15 @@ function buildAwContext() {
};
}

module.exports = { buildAwContext, buildWorkflowCallId, resolveItemContext, parseInboundAwContext, readInboundAwContext, readInboundWorkQueueAssignment, readWorkQueueAssignment, normalizeWorkQueueContext };
module.exports = {
buildAwContext,
buildWorkflowCallId,
resolveItemContext,
parseInboundAwContext,
readInboundAwContext,
readInboundWorkQueueAssignment,
readWorkQueueAssignment,
normalizeWorkQueueContext,
normalizeLegacyWorkQueueContext,
resolveWorkQueueRuntime,
};
Loading
Loading