Merge pull request #486 from Opencode-DCP/dev

fix: skip malformed messages in DCP hooks

Author: Tarquinen

lib/compress/search.ts

@@ -1,6 +1,7 @@
 import type { SessionState, WithParts } from "../state"
 import { formatBlockRef, parseBoundaryId } from "../message-ids"
 import { isIgnoredUserMessage } from "../messages/query"
+import { filterProcessableMessages } from "../messages/shape"
 import { countAllMessageTokens } from "../token-utils"
 import type { BoundaryReference, SearchContext, SelectionResolution } from "./types"
 
@@ -9,8 +10,7 @@ export async function fetchSessionMessages(client: any, sessionId: string): Prom
         path: { id: sessionId },
     })
 
-    const payload = (response?.data || response) as WithParts[]
-    return Array.isArray(payload) ? payload : []
+    return filterProcessableMessages(response?.data || response)
 }
 
 export function buildSearchContext(state: SessionState, rawMessages: WithParts[]): SearchContext {

lib/hooks.ts

@@ -22,6 +22,7 @@ import {
     consumeCompressionStart,
     resolveCompressionDuration,
 } from "./compress/timing"
+import { filterProcessableMessages } from "./messages/shape"
 import {
     applyPendingManualTrigger,
     handleContextCommand,
@@ -103,41 +104,49 @@ export function createChatMessageTransformHandler(
     hostPermissions: HostPermissionSnapshot,
 ) {
     return async (input: {}, output: { messages: WithParts[] }) => {
-        await checkSession(client, state, logger, output.messages, config.manualMode.enabled)
+        const messages = filterProcessableMessages(output.messages)
+        if (messages.length !== output.messages.length) {
+            logger.warn("Skipping messages with unexpected shape during chat transform", {
+                received: output.messages.length,
+                usable: messages.length,
+            })
+        }
+
+        await checkSession(client, state, logger, messages, config.manualMode.enabled)
 
-        syncCompressPermissionState(state, config, hostPermissions, output.messages)
+        syncCompressPermissionState(state, config, hostPermissions, messages)
 
         if (state.isSubAgent && !config.experimental.allowSubAgents) {
             return
         }
 
         stripHallucinations(output.messages)
-        cacheSystemPromptTokens(state, output.messages)
-        assignMessageRefs(state, output.messages)
-        syncCompressionBlocks(state, logger, output.messages)
-        syncToolCache(state, config, logger, output.messages)
-        buildToolIdList(state, output.messages)
-        prune(state, logger, config, output.messages)
+        cacheSystemPromptTokens(state, messages)
+        assignMessageRefs(state, messages)
+        syncCompressionBlocks(state, logger, messages)
+        syncToolCache(state, config, logger, messages)
+        buildToolIdList(state, messages)
+        prune(state, logger, config, messages)
         await injectExtendedSubAgentResults(
             client,
             state,
             logger,
-            output.messages,
+            messages,
             config.experimental.allowSubAgents,
         )
-        const compressionPriorities = buildPriorityMap(config, state, output.messages)
+        const compressionPriorities = buildPriorityMap(config, state, messages)
         prompts.reload()
         injectCompressNudges(
             state,
             config,
             logger,
-            output.messages,
+            messages,
             prompts.getRuntimePrompts(),
             compressionPriorities,
         )
-        injectMessageIds(state, config, output.messages, compressionPriorities)
-        applyPendingManualTrigger(state, output.messages, logger)
-        stripStaleMetadata(output.messages)
+        injectMessageIds(state, config, messages, compressionPriorities)
+        applyPendingManualTrigger(state, messages, logger)
+        stripStaleMetadata(messages)
 
         if (state.sessionId) {
             await logger.saveContext(state.sessionId, output.messages)
@@ -165,7 +174,7 @@ export function createCommandExecuteHandler(
             const messagesResponse = await client.session.messages({
                 path: { id: input.sessionID },
             })
-            const messages = (messagesResponse.data || messagesResponse) as WithParts[]
+            const messages = filterProcessableMessages(messagesResponse.data || messagesResponse)
 
             await ensureSessionInitialized(
                 client,

lib/messages/inject/subagent-results.ts

@@ -1,5 +1,6 @@
 import type { Logger } from "../../logger"
 import type { SessionState, WithParts } from "../../state"
+import { filterProcessableMessages } from "../shape"
 import {
     buildSubagentResultText,
     getSubAgentId,
@@ -12,8 +13,7 @@ async function fetchSubAgentMessages(client: any, sessionId: string): Promise<Wi
         path: { id: sessionId },
     })
 
-    const payload = (response?.data || response) as WithParts[]
-    return Array.isArray(payload) ? payload : []
+    return filterProcessableMessages(response?.data || response)
 }
 
 export const injectExtendedSubAgentResults = async (

lib/messages/query.ts

@@ -1,5 +1,6 @@
 import type { PluginConfig } from "../config"
 import type { WithParts } from "../state"
+import { isMessageWithInfo } from "./shape"
 
 export const getLastUserMessage = (
     messages: WithParts[],
@@ -8,6 +9,9 @@ export const getLastUserMessage = (
     const start = startIndex ?? messages.length - 1
     for (let i = start; i >= 0; i--) {
         const msg = messages[i]
+        if (!isMessageWithInfo(msg)) {
+            continue
+        }
         if (msg.info.role === "user" && !isIgnoredUserMessage(msg)) {
             return msg
         }
@@ -16,6 +20,10 @@ export const getLastUserMessage = (
 }
 
 export const messageHasCompress = (message: WithParts): boolean => {
+    if (!isMessageWithInfo(message)) {
+        return false
+    }
+
     if (message.info.role !== "assistant") {
         return false
     }
@@ -28,6 +36,10 @@ export const messageHasCompress = (message: WithParts): boolean => {
 }
 
 export const isIgnoredUserMessage = (message: WithParts): boolean => {
+    if (!isMessageWithInfo(message)) {
+        return false
+    }
+
     if (message.info.role !== "user") {
         return false
     }
@@ -47,6 +59,10 @@ export const isIgnoredUserMessage = (message: WithParts): boolean => {
 }
 
 export function isProtectedUserMessage(config: PluginConfig, message: WithParts): boolean {
+    if (!isMessageWithInfo(message)) {
+        return false
+    }
+
     return (
         config.compress.mode === "message" &&
         config.compress.protectUserMessages &&

lib/messages/shape.ts

@@ -0,0 +1,33 @@
+import type { WithParts } from "../state"
+
+export function isMessageWithInfo(message: unknown): message is WithParts {
+    if (!message || typeof message !== "object") {
+        return false
+    }
+
+    const info = (message as any).info
+    const parts = (message as any).parts
+    if (!info || typeof info !== "object") {
+        return false
+    }
+
+    return (
+        typeof info.id === "string" &&
+        info.id.length > 0 &&
+        typeof info.sessionID === "string" &&
+        info.sessionID.length > 0 &&
+        (info.role === "user" || info.role === "assistant") &&
+        info.time &&
+        typeof info.time === "object" &&
+        typeof info.time.created === "number" &&
+        Array.isArray(parts)
+    )
+}
+
+export function filterProcessableMessages(messages: unknown): WithParts[] {
+    if (!Array.isArray(messages)) {
+        return []
+    }
+
+    return messages.filter(isMessageWithInfo)
+}

lib/state/utils.ts

@@ -6,9 +6,14 @@ import type {
     WithParts,
 } from "./types"
 import { isIgnoredUserMessage, messageHasCompress } from "../messages/query"
+import { isMessageWithInfo } from "../messages/shape"
 import { countTokens } from "../token-utils"
 
 export const isMessageCompacted = (state: SessionState, msg: WithParts): boolean => {
+    if (!isMessageWithInfo(msg)) {
+        return false
+    }
+
     if (msg.info.time.created < state.lastCompaction) {
         return true
     }
@@ -58,6 +63,9 @@ export async function isSubAgentSession(client: any, sessionID: string): Promise
 export function findLastCompactionTimestamp(messages: WithParts[]): number {
     for (let i = messages.length - 1; i >= 0; i--) {
         const msg = messages[i]
+        if (!isMessageWithInfo(msg)) {
+            continue
+        }
         if (msg.info.role === "assistant" && msg.info.summary === true) {
             return msg.info.time.created
         }
@@ -68,6 +76,9 @@ export function findLastCompactionTimestamp(messages: WithParts[]): number {
 export function countTurns(state: SessionState, messages: WithParts[]): number {
     let turnCount = 0
     for (const msg of messages) {
+        if (!isMessageWithInfo(msg)) {
+            continue
+        }
         if (isMessageCompacted(state, msg)) {
             continue
         }

tests/hooks-permission.test.ts

@@ -113,6 +113,44 @@ test("chat message transform strips hallucinated tags even when compress is deni
     assert.equal((output.messages[0]?.parts[0] as any).text, "alpha  omega")
 })
 
+test("chat message transform ignores messages without info instead of crashing", async () => {
+    const state = createSessionState()
+    const logger = new Logger(false)
+    const config = buildConfig("deny")
+    const handler = createChatMessageTransformHandler(
+        { session: { get: async () => ({}) } } as any,
+        state,
+        logger,
+        config,
+        {
+            reload() {},
+            getRuntimePrompts() {
+                return {} as any
+            },
+        } as any,
+        { global: undefined, agents: {} },
+    )
+    const output = {
+        messages: [
+            {
+                role: "user",
+                time: 1,
+                parts: [
+                    {
+                        type: "text",
+                        text: "Carica le skill di laravel",
+                    },
+                ],
+            } as any,
+        ],
+    }
+
+    await handler({}, output as any)
+
+    assert.equal(state.sessionId, null)
+    assert.equal(output.messages.length, 1)
+})
+
 test("command execute exits after effective permission resolves to deny", async () => {
     let sessionMessagesCalls = 0
     const output = { parts: [] as any[] }