fix: prevent cross-task contamination and mutation side effects by deep-cloning persisted and incoming task data

Author: JakubWorek

src/server/result_manager.ts

@@ -9,6 +9,18 @@ import { ServerCallContext } from './context.js';
 import { AgentExecutionEvent, assertUnreachableEvent } from './events/execution_event_bus.js';
 import { TaskStore } from './store.js';
 
+/**
+ * Tracks the in-flight task/message state for a single A2A request and
+ * persists updates to the {@link TaskStore}.
+ *
+ * Mutation safety: every external object handed to this class (event
+ * payloads, user messages) is deep-cloned via `structuredClone` before
+ * being stored, and every object stored back into the task is likewise a
+ * fresh clone. This isolates `ResultManager`'s internal state from
+ * caller-side mutations of the same event objects (and vice versa). The
+ * `TaskStore.load` / `TaskStore.save` boundary clones independently, so
+ * this class is safe to combine with stores that share references.
+ */
 export class ResultManager {
   private readonly taskStore: TaskStore;
   private readonly serverCallContext: ServerCallContext;
@@ -23,7 +35,9 @@ export class ResultManager {
   }
 
   public setContext(latestUserMessage: Message): void {
-    this.latestUserMessage = latestUserMessage;
+    // Clone so a caller mutating the message later (or reusing the object
+    // across calls) can't perturb our internal copy.
+    this.latestUserMessage = structuredClone(latestUserMessage);
   }
 
   /**
@@ -33,7 +47,9 @@ export class ResultManager {
   public async processEvent(event: AgentExecutionEvent): Promise<void> {
     switch (event.kind) {
       case 'message': {
-        this.finalMessageResult = event.data;
+        // Final-result messages may be returned to callers verbatim, so
+        // store a defensive copy.
+        this.finalMessageResult = structuredClone(event.data);
         // If a message is received, it's usually the final result,
         // but we continue processing to ensure task state (if any) is also saved.
         // The ExecutionEventQueue will stop after a message event.
@@ -50,52 +66,66 @@ export class ResultManager {
         // Unlike status/artifact updates, receiving a Task event with no
         // prior persisted task is the normal create-flow, so we load
         // directly rather than going through `ensureTaskLoaded` (which
-        // warns on misses).
-        if (!this.currentTask && taskEvent.id) {
+        // warns on misses). We also re-load if the in-memory task is for
+        // a different id, otherwise we'd lose the persisted state for the
+        // new task id.
+        if ((!this.currentTask || this.currentTask.id !== taskEvent.id) && taskEvent.id) {
           const loaded = await this.taskStore.load(taskEvent.id, this.serverCallContext);
           if (loaded) {
             this.currentTask = loaded;
+          } else if (this.currentTask && this.currentTask.id !== taskEvent.id) {
+            // The previously-tracked task is unrelated to the incoming
+            // event; drop it so we don't accidentally merge across ids.
+            this.currentTask = undefined;
           }
         }
         const persistedTask =
           this.currentTask && this.currentTask.id === taskEvent.id ? this.currentTask : undefined;
 
-        const mergedTask: Task = { ...taskEvent };
+        // Deep-clone the incoming Task so further executor mutations or
+        // caller-side reuse of the same event object can't leak into our
+        // state.
+        const mergedTask: Task = structuredClone(taskEvent);
 
         if (persistedTask) {
           // Preserve persisted history when the incoming Task event omits it.
           // If the incoming Task event carries its own history, treat it as
           // authoritative (the executor is responsible for what gets persisted
           // per §3.7).
           if ((!mergedTask.history || mergedTask.history.length === 0) && persistedTask.history) {
-            mergedTask.history = [...persistedTask.history];
+            mergedTask.history = structuredClone(persistedTask.history);
           }
 
           // Merge artifacts: keep persisted artifacts and overlay any incoming
           // ones (matched by artifactId). Incoming wins for collisions; new
           // ones are appended.
           mergedTask.artifacts = this.mergeArtifacts(persistedTask.artifacts, taskEvent.artifacts);
 
-          // Merge metadata, incoming wins on key collisions.
+          // Merge metadata, incoming wins on key collisions. structuredClone
+          // each half so nested values can't be shared with either source.
           if (persistedTask.metadata || taskEvent.metadata) {
             mergedTask.metadata = {
-              ...(persistedTask.metadata ?? {}),
-              ...(taskEvent.metadata ?? {}),
+              ...structuredClone(persistedTask.metadata ?? {}),
+              ...structuredClone(taskEvent.metadata ?? {}),
             };
           }
         }
 
         this.currentTask = mergedTask;
 
         // Ensure the latest user message is in history if not already present.
+        // `latestUserMessage` was already cloned in `setContext`, but clone
+        // again so the same reference can't end up shared between the
+        // history array and the `latestUserMessage` slot if the same
+        // `ResultManager` is reused for multiple task events.
         if (this.latestUserMessage) {
           if (
             !this.currentTask.history?.find(
               (msg) => msg.messageId === this.latestUserMessage?.messageId
             )
           ) {
             this.currentTask.history = [
-              this.latestUserMessage,
+              structuredClone(this.latestUserMessage),
               ...(this.currentTask.history || []),
             ];
           }
@@ -131,12 +161,14 @@ export class ResultManager {
     await this.ensureTaskLoaded(updateEvent.taskId, 'status update');
 
     if (this.currentTask && this.currentTask.id === updateEvent.taskId) {
-      this.currentTask.status = updateEvent.status;
+      // Clone the incoming status (and its nested message) so caller-side
+      // mutation of the original event payload can't drift our state.
+      this.currentTask.status = structuredClone(updateEvent.status);
       const update = updateEvent.status?.message;
       if (update) {
         // Add message to history if not already present
         if (!this.currentTask.history?.find((msg) => msg.messageId === update.messageId)) {
-          this.currentTask.history = [...(this.currentTask.history || []), update];
+          this.currentTask.history = [...(this.currentTask.history || []), structuredClone(update)];
         }
       }
       await this.saveCurrentTask();
@@ -158,22 +190,27 @@ export class ResultManager {
       );
       if (existingArtifactIndex !== -1) {
         if (artifactEvent.append) {
-          // Basic append logic, assuming parts are compatible
-          // More sophisticated merging might be needed for specific part types
+          // Basic append logic, assuming parts are compatible.
+          // Clone incoming parts/metadata so the persisted artifact owns
+          // its own deep copies and the event payload can be reused
+          // safely by the executor.
           const existingArtifact = this.currentTask.artifacts[existingArtifactIndex];
-          existingArtifact.parts = [...(existingArtifact.parts || []), ...(artifact.parts || [])];
+          existingArtifact.parts = [
+            ...(existingArtifact.parts || []),
+            ...structuredClone(artifact.parts || []),
+          ];
           if (artifact.description) existingArtifact.description = artifact.description;
           if (artifact.name) existingArtifact.name = artifact.name;
           if (artifact.metadata)
             existingArtifact.metadata = {
               ...existingArtifact.metadata,
-              ...artifact.metadata,
+              ...structuredClone(artifact.metadata),
             };
         } else {
-          this.currentTask.artifacts[existingArtifactIndex] = artifact;
+          this.currentTask.artifacts[existingArtifactIndex] = structuredClone(artifact);
         }
       } else {
-        this.currentTask.artifacts.push(artifact);
+        this.currentTask.artifacts.push(structuredClone(artifact));
       }
       await this.saveCurrentTask();
     }
@@ -184,28 +221,34 @@ export class ResultManager {
    * are retained and overlaid by any incoming artifact with the same id;
    * artifacts only present in the incoming list are appended. Order is
    * preserved (persisted first, then any newly-introduced incoming artifacts).
+   *
+   * Every artifact in the returned array is a fresh deep copy so subsequent
+   * in-place mutations (e.g. by `applyArtifactUpdate`) can't leak into the
+   * original `persisted` / `incoming` arrays the caller still holds.
    */
   private mergeArtifacts(
     persisted: Artifact[] | undefined,
     incoming: Artifact[] | undefined
   ): Artifact[] {
     if (!persisted || persisted.length === 0) {
-      return incoming ? [...incoming] : [];
+      return incoming ? structuredClone(incoming) : [];
     }
     if (!incoming || incoming.length === 0) {
-      return [...persisted];
+      return structuredClone(persisted);
     }
 
     const incomingById = new Map<string, Artifact>();
     for (const art of incoming) {
       incomingById.set(art.artifactId, art);
     }
 
-    const merged: Artifact[] = persisted.map((art) => incomingById.get(art.artifactId) ?? art);
+    const merged: Artifact[] = persisted.map((art) =>
+      structuredClone(incomingById.get(art.artifactId) ?? art)
+    );
     const seenIds = new Set(persisted.map((art) => art.artifactId));
     for (const art of incoming) {
       if (!seenIds.has(art.artifactId)) {
-        merged.push(art);
+        merged.push(structuredClone(art));
       }
     }
     return merged;
@@ -220,6 +263,11 @@ export class ResultManager {
   /**
    * Gets the final result, which could be a Message or a Task.
    * This should be called after the event stream has been fully processed.
+   *
+   * Returns a reference to the internally-tracked object (not a clone) so
+   * that downstream callers like `_applyHistoryLengthSemantics` can apply
+   * in-place edits. Safe because `ResultManager` instances are scoped to a
+   * single request and discarded immediately after this is called.
    * @returns The final Message or the current Task.
    */
   public getFinalResult(): Message | Task | undefined {
@@ -232,6 +280,9 @@ export class ResultManager {
   /**
    * Gets the task currently being managed by this ResultManager instance.
    * This task could be one that was started with or one created during agent execution.
+   *
+   * Returns the internal reference (see {@link getFinalResult} for the
+   * rationale).
    * @returns The current Task or undefined if no task is active.
    */
   public getCurrentTask(): Task | undefined {