Skip to content

fix: guard goal continuation across compaction epochs #61

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
harryzhou2000 wants to merge 1 commit into from
Closed

fix: guard goal continuation across compaction epochs #61

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
163 changes: 150 additions & 13 deletions src/goal-plugin.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ function legacyHomeStateFilePath(env = process.env) {
return join(homeBase(env), ".opencode-goal-plugin", "state.json")
}
const MAX_HISTORY_ENTRIES = 20
const MAX_STALLED_COMPACTIONS = 2
// Marks a plugin-synthesized parent wake so the receiving pass knows it is
// re-examining an assistant turn that has already been scored.
const CHILD_WAKE_EVENT_FLAG = Symbol.for("opencode-goal-plugin.childWake")
Expand Down Expand Up @@ -1104,6 +1105,10 @@ function resetGoalBudget(goal) {
goal.formatFailures = 0
goal.lastAssistantMessageID = ""
goal.continuationClaim = null
goal.compactionEpoch = 0
goal.stalledCompactions = 0
goal.lastCompactionEventID = ""
goal.compactionSourceAssistantMessageID = ""
goal.skipNextTerminalCheck = false
goal.history = [...(goal.history || [])].slice(-MAX_HISTORY_ENTRIES)
}
Expand Down Expand Up @@ -1433,15 +1438,30 @@ function normalizePersistedGoal(rawGoal) {
stopReason: typeof rawGoal.stopReason === "string" ? rawGoal.stopReason : "",
promptFailures: toNonNegativeInteger(rawGoal.promptFailures),
formatFailures: toNonNegativeInteger(rawGoal.formatFailures),
compactionEpoch: toNonNegativeInteger(rawGoal.compactionEpoch),
stalledCompactions: toNonNegativeInteger(rawGoal.stalledCompactions),
lastCompactionEventID:
typeof rawGoal.lastCompactionEventID === "string" &&
rawGoal.lastCompactionEventID.length <= MAX_GOAL_META_LENGTH
? rawGoal.lastCompactionEventID
: "",
compactionSourceAssistantMessageID:
typeof rawGoal.compactionSourceAssistantMessageID === "string" &&
rawGoal.compactionSourceAssistantMessageID.length <= MAX_GOAL_META_LENGTH
? rawGoal.compactionSourceAssistantMessageID
: "",
executionContext: normalizeExecutionContext(rawGoal.executionContext),
continuationClaim:
isPlainObject(rawGoal.continuationClaim) &&
typeof rawGoal.continuationClaim.runId === "string" &&
rawGoal.continuationClaim.runId.length <= MAX_GOAL_META_LENGTH &&
Number.isSafeInteger(rawGoal.continuationClaim.compactionEpoch) &&
rawGoal.continuationClaim.compactionEpoch >= 0 &&
typeof rawGoal.continuationClaim.sourceAssistantMessageID === "string" &&
rawGoal.continuationClaim.sourceAssistantMessageID.length <= MAX_GOAL_META_LENGTH
? {
runId: rawGoal.continuationClaim.runId,
compactionEpoch: rawGoal.continuationClaim.compactionEpoch,
sourceAssistantMessageID: rawGoal.continuationClaim.sourceAssistantMessageID,
}
: null,
Expand Down Expand Up @@ -2635,7 +2655,42 @@ function systemBlockContainsGoal(block, goalId) {
}

function findLatestAssistantMessage(messages) {
return [...(messages || [])].reverse().find((message) => messageRole(message) === "assistant") || null
return [...(messages || [])]
.reverse()
.find(
(message) =>
messageRole(message) === "assistant" && !isCompactionAssistantMessage(message),
) || null
}

function isCompactionAssistantMessage(message) {
if (messageRole(message) !== "assistant") return false
const info = isPlainObject(message?.info) ? message.info : message
return (
info?.summary === true ||
info?.agent === "compaction" ||
info?.mode === "compaction" ||
message?.agent === "compaction" ||
message?.mode === "compaction"
)
}

function compactionEventIdentity(event) {
const candidates = [
event?.id,
event?.properties?.compactionID,
event?.properties?.summaryID,
event?.properties?.messageID,
event?.properties?.id,
event?.data?.compactionID,
event?.data?.summaryID,
event?.data?.messageID,
event?.data?.id,
]
const identity = candidates.find(
(candidate) => typeof candidate === "string" && candidate.length > 0,
)
return identity && identity.length <= MAX_GOAL_META_LENGTH ? identity : ""
}

function messageParentID(message) {
Expand Down Expand Up @@ -2771,6 +2826,7 @@ function continuationSnapshot(messages, ownedMessages = currentRuntime().ownedPl
.reverse()
.find((message) =>
(messageRole(message) === "assistant" || messageRole(message) === "user") &&
!isCompactionAssistantMessage(message) &&
!isPluginGeneratedMessage(message, ownedMessages),
)
return {
Expand Down Expand Up @@ -2939,6 +2995,10 @@ function buildGoalState(sessionID, condition, options, meta = {}, lastStatus = "
stopReason: "",
promptFailures: 0,
formatFailures: 0,
compactionEpoch: 0,
stalledCompactions: 0,
lastCompactionEventID: "",
compactionSourceAssistantMessageID: "",
executionContext: normalizeExecutionContext(
meta.executionContext || currentRuntime().sessionExecutionContexts.get(sessionID),
),
Expand Down Expand Up @@ -4397,18 +4457,19 @@ async function createGoalPlugin({ client, directory } = {}, pluginOptions = {})
sessionID,
goalID,
runID,
compactionEpoch,
baselineMessages,
{ refreshMessages = false } = {},
) => {
const goalBeforeRefresh = activeGoal(sessionID, goalID, runID)
if (!goalBeforeRefresh) return null
if (!goalBeforeRefresh || goalBeforeRefresh.compactionEpoch !== compactionEpoch) return null
const hostMessages = refreshMessages
? await sessionApi.messages(sessionID, {
limit: goalBeforeRefresh.options.maxRecentMessages,
})
: baselineMessages
const goal = activeGoal(sessionID, goalID, runID)
if (!goal) return null
if (!goal || goal.compactionEpoch !== compactionEpoch) return null
const messages = Array.isArray(hostMessages)
? hostMessages.slice(-goal.options.maxRecentMessages)
: []
Expand Down Expand Up @@ -4523,12 +4584,13 @@ async function createGoalPlugin({ client, directory } = {}, pluginOptions = {})
const sourceAssistantMessageID = refreshed.latestAssistantID || "<no-assistant>"
if (
goal.continuationClaim?.runId === runID &&
goal.continuationClaim?.compactionEpoch === compactionEpoch &&
goal.continuationClaim?.sourceAssistantMessageID === sourceAssistantMessageID
) {
return null
}

goal.continuationClaim = { runId: runID, sourceAssistantMessageID }
goal.continuationClaim = { runId: runID, compactionEpoch, sourceAssistantMessageID }
const claimPersisted = await persist(sessionID)
if (!claimPersisted && persistenceOptions.persistState) {
goal.continuationClaim = null
Expand All @@ -4545,7 +4607,13 @@ async function createGoalPlugin({ client, directory } = {}, pluginOptions = {})
})
return null
}
return goal
// Let an already-published compaction event invalidate this claim before
// the caller enters promptAsync. The final epoch check is the atomic edge:
// a claim is valid only while its context epoch is still current.
await Promise.resolve()
return activeGoal(sessionID, goalID, runID)?.compactionEpoch === compactionEpoch
? goal
: null
}

const retireCompletedCommandTurnOnIdle = async (sessionID, messageLimit) => {
Expand Down Expand Up @@ -5328,22 +5396,56 @@ async function createGoalPlugin({ client, directory } = {}, pluginOptions = {})
if (event?.type === "session.compacted") {
const sessionID = getSessionID(event)
const goal = goalStates.get(sessionID)
if (!goal) return
if (!goal || goal.stopped) return
const identity = compactionEventIdentity(event)
if (identity && identity === goal.lastCompactionEventID) return
if (identity) goal.lastCompactionEventID = identity

goal.compactionEpoch += 1
goal.stalledCompactions += 1
goal.compactionSourceAssistantMessageID =
goal.continuationClaim?.runId === goal.runId
? goal.continuationClaim.sourceAssistantMessageID
: ""
goal.messageIDs = new Set()
goal.totalTokens = 0
// Compaction rewrites the context: a continuation claim for a
// pre-compaction source turn must not suppress the post-compaction
// continuation (the recent tail can still end on the same assistant
// message, which would otherwise stall the goal loop until the user
// nudges it).
// Compaction rewrites the context. The epoch-scoped claim lets the same
// retained assistant source continue once in the new epoch without
// allowing duplicate idle delivery to continue it twice.
goal.continuationClaim = null

// An idle handler can already have persisted its source claim when the
// compaction lands. Abort its cooldown and release the per-session guard;
// the epoch checks around promptAsync prevent that stale handler from
// sending while allowing the post-compaction idle to start immediately.
currentRuntime().continuationControllers.get(sessionID)?.abort()
currentRuntime().continuationControllers.delete(sessionID)
activeContinues.delete(sessionID)

if (goal.stalledCompactions >= MAX_STALLED_COMPACTIONS) {
await pauseActiveGoal(sessionID, {
stopReason: "stalled compaction",
status: `Goal paused after ${goal.stalledCompactions} compactions without a productive assistant or tool turn.`,
history: `Paused after ${goal.stalledCompactions} compactions without productive non-compaction work.`,
})
if (typeof client?.session?.abort === "function") {
try {
await sessionApi.abort(sessionID)
} catch (error) {
await logPluginError(client, "Failed to abort a stalled compaction loop", error)
}
}
return
}
await persist(sessionID)
return
}

if (event?.type === "message.updated") {
const message = messageInfoFromEvent(event)
if (!message) return
const messageEnvelope =
event?.properties?.message || event?.data?.message || message

const currentMessageID = messageID(message)
if (!currentMessageID) return
Expand Down Expand Up @@ -5395,13 +5497,27 @@ async function createGoalPlugin({ client, directory } = {}, pluginOptions = {})

if (
messageRole(message) === "assistant" &&
!isCompactionAssistantMessage(messageEnvelope) &&
currentMessageID !== goal.compactionSourceAssistantMessageID &&
currentOutputTokens > previousOutputTokens &&
runtime.suppressedCommandAssistants.get(currentMessageID) !== currentSessionID
) {
goal.lastProgressAt = Date.now()
changed = true
}

if (
messageRole(message) === "assistant" &&
!isCompactionAssistantMessage(messageEnvelope) &&
currentMessageID !== goal.compactionSourceAssistantMessageID &&
(currentOutputTokens > previousOutputTokens || messageHasToolCall(messageEnvelope)) &&
goal.stalledCompactions > 0
) {
goal.stalledCompactions = 0
goal.compactionSourceAssistantMessageID = ""
changed = true
}

if (changed) await persist(messageSessionID(message))
return
}
Expand Down Expand Up @@ -5497,10 +5613,12 @@ async function createGoalPlugin({ client, directory } = {}, pluginOptions = {})
if (!goal || goal.stopped || activeContinues.has(sessionID)) return
const goalID = goal.goalId
const runID = goal.runId
const compactionEpoch = goal.compactionEpoch

const continueToken = randomUUID()
const continueController = new AbortController()
let claimedSourceAssistantMessageID = ""
let claimedCompactionEpoch = -1
activeContinues.set(sessionID, continueToken)
currentRuntime().continuationControllers.set(sessionID, continueController)
try {
Expand All @@ -5513,7 +5631,10 @@ async function createGoalPlugin({ client, directory } = {}, pluginOptions = {})
? hostMessages.slice(-goal.options.maxRecentMessages)
: []
const activeGoalAfterMessages = activeGoal(sessionID, goalID, runID)
if (!activeGoalAfterMessages) return
if (
!activeGoalAfterMessages ||
activeGoalAfterMessages.compactionEpoch !== compactionEpoch
) return
if (!activeGoalAfterMessages.executionContext) {
activeGoalAfterMessages.executionContext = findLatestExecutionContext(messages)
}
Expand All @@ -5528,7 +5649,11 @@ async function createGoalPlugin({ client, directory } = {}, pluginOptions = {})
latestAssistantID && latestAssistantID === activeGoalAfterMessages.lastAssistantMessageID
const activationBoundary =
currentRuntime().suppressedCommandAssistants.get(latestAssistantID) === sessionID ||
activeGoalAfterMessages.skipNextTerminalCheck === true
activeGoalAfterMessages.skipNextTerminalCheck === true ||
(
activeGoalAfterMessages.compactionSourceAssistantMessageID &&
activeGoalAfterMessages.compactionSourceAssistantMessageID === latestAssistantID
)
activeGoalAfterMessages.skipNextTerminalCheck = false

if (!activationBoundary && latestText && (!assistantRepeated || assistantChanged)) {
Expand All @@ -5555,6 +5680,7 @@ async function createGoalPlugin({ client, directory } = {}, pluginOptions = {})
const sourceAssistantMessageID = latestAssistantID || "<no-assistant>"
if (
activeGoalAfterMessages.continuationClaim?.runId === runID &&
activeGoalAfterMessages.continuationClaim?.compactionEpoch === compactionEpoch &&
activeGoalAfterMessages.continuationClaim?.sourceAssistantMessageID ===
sourceAssistantMessageID
) {
Expand Down Expand Up @@ -5801,6 +5927,7 @@ async function createGoalPlugin({ client, directory } = {}, pluginOptions = {})
sessionID,
goalID,
runID,
compactionEpoch,
messages,
)
if (!claimedGoal) return
Expand Down Expand Up @@ -6014,12 +6141,16 @@ async function createGoalPlugin({ client, directory } = {}, pluginOptions = {})
sessionID,
goalID,
runID,
compactionEpoch,
messages,
{ refreshMessages: cooldownWaited },
)
if (!activeGoalBeforePrompt) return
claimedSourceAssistantMessageID =
activeGoalBeforePrompt.continuationClaim?.sourceAssistantMessageID || ""
claimedCompactionEpoch =
activeGoalBeforePrompt.continuationClaim?.compactionEpoch ?? -1
if (claimedCompactionEpoch !== activeGoalBeforePrompt.compactionEpoch) return

const budgetWrapup = budgetWrapupNeeded(activeGoalBeforePrompt)
if (budgetWrapup) {
Expand Down Expand Up @@ -6115,6 +6246,8 @@ async function createGoalPlugin({ client, directory } = {}, pluginOptions = {})
const activeGoalAfterPrompt = currentGoal(sessionID, goalID, runID)
const message = `Auto-continue failed: ${response.error.name || "unknown error"}`
if (
activeGoalAfterPrompt?.continuationClaim?.compactionEpoch ===
claimedCompactionEpoch &&
activeGoalAfterPrompt?.continuationClaim?.sourceAssistantMessageID ===
claimedSourceAssistantMessageID
) {
Expand All @@ -6133,6 +6266,8 @@ async function createGoalPlugin({ client, directory } = {}, pluginOptions = {})
} else {
const activeGoalAfterPrompt = currentGoal(sessionID, goalID, runID)
if (
activeGoalAfterPrompt?.continuationClaim?.compactionEpoch ===
claimedCompactionEpoch &&
activeGoalAfterPrompt?.continuationClaim?.sourceAssistantMessageID ===
claimedSourceAssistantMessageID
) {
Expand Down Expand Up @@ -6164,6 +6299,8 @@ async function createGoalPlugin({ client, directory } = {}, pluginOptions = {})
if (activeGoalAfterError) {
if (
claimedSourceAssistantMessageID &&
activeGoalAfterError.continuationClaim?.compactionEpoch ===
claimedCompactionEpoch &&
activeGoalAfterError.continuationClaim?.sourceAssistantMessageID ===
claimedSourceAssistantMessageID
) {
Expand Down
Loading
Loading