refactor(task-graph): apply PR #448 review feedback

Four small fixes flagged by Copilot review on the Workflow decomposition:

1. WorkflowEventBridge.detach() now also tears down the streaming
   subscription and clears _streamingUnsub. Without this, a graph swap
   or reset() during a run would leak the old graph's streaming
   handlers wired to this workflow's events emitter.

2. runLoopAutoConnect / consumePendingConnect now return the error
   message instead of just logging. Callers (Workflow#run for the loop-
   builder branch, Workflow#autoConnectLoopTask, LoopBuilderContext#
   finalizeAndReturn) propagate the error onto the parent workflow's
   builder via the new WorkflowBuilder#setError(message). This restores
   parity with the pre-refactor behavior where loop auto-connect
   failures surfaced on Workflow.error, not just in the logger.

3. Workflow.ts switches `import { TaskOutputRepository }` to
   `import type` since it's only used in type positions.

4. WorkflowPipe.connect() now uses the canonical DATAFLOW_ALL_PORTS
   constant instead of the hard-coded "*" sentinel.

Author: sroussey

packages/task-graph/src/task-graph/LoopBuilderContext.ts

@@ -19,13 +19,21 @@ export interface PendingLoopConnect {
 /**
  * Runs deferred auto-connect for a loop iterator task on the parent
  * workflow's graph. Extracted as a free function so it can be invoked
- * from both {@link LoopBuilderContext.autoConnectLoopTask} and from
+ * from both {@link LoopBuilderContext.consumePendingConnect} and from
  * the parent {@link Workflow}'s public delegate method (the parent is
  * not itself in loop-builder mode and has no context of its own).
+ *
+ * Returns the error message if auto-connect failed, otherwise undefined.
+ * On failure the iterator task is removed from the parent graph; the
+ * caller is responsible for surfacing the error onto the parent
+ * workflow's `.error` (matches the non-loop auto-connect path).
  */
-export function runLoopAutoConnect(parentGraph: TaskGraph, pending: PendingLoopConnect): void {
+export function runLoopAutoConnect(
+  parentGraph: TaskGraph,
+  pending: PendingLoopConnect
+): string | undefined {
   const { parent, iteratorTask } = pending;
-  if (parentGraph.getTargetDataflows(parent.id).length !== 0) return;
+  if (parentGraph.getTargetDataflows(parent.id).length !== 0) return undefined;
 
   const nodes = parentGraph.getTasks();
   const parentIndex = nodes.findIndex((n) => n.id === parent.id);
@@ -36,9 +44,12 @@ export function runLoopAutoConnect(parentGraph: TaskGraph, pending: PendingLoopC
 
   const result = autoConnect(parentGraph, parent, iteratorTask, { earlierTasks });
   if (result.error) {
-    getLogger().error(result.error + " Task not added.");
+    const message = result.error + " Task not added.";
+    getLogger().error(message);
     parentGraph.removeTask(iteratorTask.id);
+    return message;
   }
+  return undefined;
 }
 
 /**
@@ -66,18 +77,28 @@ export class LoopBuilderContext {
     this.iteratorTask.validateAcyclic();
   }
 
-  /** Runs auto-connect for the pending loop connect (if any), then clears it. */
-  public consumePendingConnect(): void {
+  /**
+   * Runs auto-connect for the pending loop connect (if any), then clears it.
+   * Returns the error message if auto-connect failed, otherwise undefined,
+   * so the caller can propagate the failure to the parent workflow's `.error`.
+   */
+  public consumePendingConnect(): string | undefined {
     const pending = this.pendingLoopConnect;
-    if (!pending) return;
-    runLoopAutoConnect(this.parent.graph, pending);
+    if (!pending) return undefined;
+    const error = runLoopAutoConnect(this.parent.graph, pending);
     this.pendingLoopConnect = undefined;
+    return error;
   }
 
-  /** Finalizes the template and returns the parent workflow. */
+  /**
+   * Finalizes the template and returns the parent workflow. Any deferred
+   * auto-connect error is surfaced onto the parent's `.error` to match the
+   * non-loop auto-connect path.
+   */
   public finalizeAndReturn(childGraph: TaskGraph): Workflow {
     this.finalizeTemplate(childGraph);
-    this.consumePendingConnect();
+    const error = this.consumePendingConnect();
+    if (error) this.parent.builder.setError(error);
     return this.parent;
   }
 }

packages/task-graph/src/task-graph/Workflow.ts

@@ -6,7 +6,7 @@
 
 import type { EventParameters } from "@workglow/util";
 import { EventEmitter, ServiceRegistry } from "@workglow/util";
-import { TaskOutputRepository } from "../storage/TaskOutputRepository";
+import type { TaskOutputRepository } from "../storage/TaskOutputRepository";
 import type { ConditionFn } from "../task/ConditionalTask";
 import { GraphAsTask } from "../task/GraphAsTask";
 import type { ITask, ITaskConstructor } from "../task/ITask";
@@ -243,7 +243,8 @@ export class Workflow<
     const loopContext = this._builder.loopContext;
     if (loopContext) {
       loopContext.finalizeTemplate(this._graph);
-      loopContext.consumePendingConnect();
+      const error = loopContext.consumePendingConnect();
+      if (error) loopContext.parent.builder.setError(error);
       return loopContext.parent.run(input as any, config) as Promise<
         PropertyArrayGraphResult<Output>
       >;
@@ -588,7 +589,8 @@ export class Workflow<
    */
   public autoConnectLoopTask(pending?: { parent: ITask; iteratorTask: ITask }): void {
     if (!pending) return;
-    runLoopAutoConnect(this._graph, pending);
+    const error = runLoopAutoConnect(this._graph, pending);
+    if (error) this._builder.setError(error);
   }
 
   /**

packages/task-graph/src/task-graph/WorkflowBuilder.ts

@@ -59,6 +59,15 @@ export class WorkflowBuilder {
     return this._loopContext;
   }
 
+  /**
+   * Surfaces an error onto this builder's error slot. Used by deferred
+   * loop-builder auto-connect (which runs from a child Workflow but reports
+   * the failure onto the parent's `.error`, matching the non-loop path).
+   */
+  public setError(message: string): void {
+    this._error = message;
+  }
+
   /** Clears pending state. Called by the facade's graph setter and reset(). */
   public resetState(): void {
     this._dataFlows = [];

packages/task-graph/src/task-graph/WorkflowEventBridge.ts

@@ -59,6 +59,11 @@ export class WorkflowEventBridge {
     graph.off("dataflow_removed", this._onChanged);
     this._entitlementUnsub?.();
     this._entitlementUnsub = undefined;
+    // Tear down any streaming subscription that's still tied to this graph.
+    // Without this, a graph swap or reset() during a run would leave the old
+    // graph's streaming handlers wired to this workflow's events emitter.
+    this._streamingUnsub?.();
+    this._streamingUnsub = undefined;
     this._attachedGraph = undefined;
   }
 

packages/task-graph/src/task-graph/WorkflowPipe.ts

@@ -9,7 +9,7 @@ import type { ITask } from "../task/ITask";
 import type { DataPorts } from "../task/TaskTypes";
 import type { PipeFunction, Taskish } from "./Conversions";
 import { ensureTask } from "./Conversions";
-import { Dataflow } from "./Dataflow";
+import { Dataflow, DATAFLOW_ALL_PORTS } from "./Dataflow";
 import type { ITaskGraph } from "./ITaskGraph";
 import type { IWorkflow } from "./IWorkflow";
 import type { CompoundMergeStrategy } from "./TaskGraphRunner";
@@ -31,7 +31,9 @@ export function connect(
   target: ITask<any, any, any>,
   workflow: IWorkflow<any, any>
 ): void {
-  workflow.graph.addDataflow(new Dataflow(source.id, "*", target.id, "*"));
+  workflow.graph.addDataflow(
+    new Dataflow(source.id, DATAFLOW_ALL_PORTS, target.id, DATAFLOW_ALL_PORTS)
+  );
 }
 
 export function pipe<A extends DataPorts, B extends DataPorts>(