feat(workflow): per-task subagent routing and a spawn-plan telemetry event

AgentDynamicWorkflow could only spawn homogeneous subagents: one
subagent_type and one model for every item.

- Input gains tasks[] ({ item, subagent_type?, model?, thinking? }) and
  defaults.subagent_type beside the homogeneous items shape; tasks and
  items cannot be combined. Each task resolves through the routing
  service, so a workflow can mix profiles, models, and thinking efforts
  and every plan carries its own provenance. Profile choices stay the
  catalog's; the tool never hard-codes them.
- SubagentSpawnPlanInput accepts an explicit thinking effort; it is part
  of the route decision fingerprint, never of the environment revision.
- The routing service emits subagent_spawn_plan_resolved (operation,
  profile/model source, policy mode and source, feature source, the two
  routing hashes, and which inputs were explicit) with no prompt content.
- Token-count expectations and wire snapshots follow the larger tool
  schema.

Author: elkaix

.changeset/dynamic-workflow-per-task.md

@@ -0,0 +1,5 @@
+---
+"@pymodel/pythinker-code": minor
+---
+
+AgentDynamicWorkflow accepts a `tasks` list where each entry sets its own subagent type, model, and thinking effort. Pass `tasks` instead of `items`, with optional `defaults.subagent_type`.

packages/agent-core-v2/src/app/telemetry/events.ts

@@ -389,6 +389,20 @@ export interface SubagentCreatedEvent {
   model?: string;
 }
 
+export interface SubagentSpawnPlanResolvedEvent {
+  operation: 'spawn' | 'fork';
+  profile_source: 'requested' | 'default' | 'fork-inherit' | 'resume-existing';
+  model_source: 'caller' | 'policy-default' | 'policy-pool' | 'policy-force' | 'fork-inherit' | 'resume-existing';
+  policy_mode: 'inherit' | 'default' | 'pool' | 'force';
+  policy_source: 'config' | 'default';
+  feature_source: 'master-env' | 'env' | 'config' | 'default';
+  routing_env_revision: string;
+  route_decision: string;
+  explicit_profile: boolean;
+  explicit_model: boolean;
+  explicit_thinking: boolean;
+}
+
 export interface McpConnectedEvent {
   server_count: number;
   total_count: number;
@@ -961,6 +975,23 @@ export const telemetryEventDefinitions = {
       model: 'Model alias the subagent binds to (secondary-model choice or inherited caller model); omitted when no binding was resolved',
     },
   }),
+  subagent_spawn_plan_resolved: defineTelemetryEvent<SubagentSpawnPlanResolvedEvent>({
+    owner: 'pythinker-code',
+    comment: 'The routing resolver produced a spawn plan for a new subagent (no prompt content).',
+    properties: {
+      operation: 'spawn or fork',
+      profile_source: 'Where the profile came from: requested, default, or fork-inherit',
+      model_source: 'Where the model came from: caller, policy-default, policy-pool, policy-force, or fork-inherit',
+      policy_mode: 'Effective subagent model policy mode',
+      policy_source: 'Whether the policy came from config or the default',
+      feature_source: 'Where the secondary-model feature state came from',
+      routing_env_revision: 'Hash of the ambient routing inputs the plan was resolved from',
+      route_decision: 'Hash of the routing environment plus the request intent',
+      explicit_profile: 'Whether the request named a subagent type',
+      explicit_model: 'Whether the request named a model',
+      explicit_thinking: 'Whether the request named a thinking effort',
+    },
+  }),
   mcp_connected: defineTelemetryEvent<McpConnectedEvent>({
     owner: 'pythinker-code',
     comment: 'MCP servers connect at session start.',

packages/agent-core-v2/src/features/dynamic_workflow/tools/agent-dynamic_workflow/agent-dynamic_workflow.ts

@@ -50,14 +50,41 @@ export const AgentDynamicWorkflowToolInputSchema = z
       .describe(
         'Map of existing subagent agent_id to the prompt used to resume that subagent. These resumed subagents are launched before new item-based subagents.',
       ),
+    defaults: z
+      .object({
+        subagent_type: z.string().trim().min(1).optional(),
+      })
+      .strict()
+      .optional()
+      .describe('Defaults applied to every entry of tasks that does not set its own value.'),
+    tasks: z
+      .array(
+        z
+          .object({
+            item: z.string().trim().min(1).describe(`Value used to fill ${PROMPT_TEMPLATE_PLACEHOLDER} for this subagent.`),
+            subagent_type: z.string().trim().min(1).optional().describe('Subagent type for this subagent; overrides defaults.subagent_type and subagent_type.'),
+            model: z.string().optional().describe('Model alias for this subagent (same vocabulary as model); overrides model.'),
+            thinking: z.string().optional().describe('Thinking effort for this subagent; overrides the model default.'),
+          })
+          .strict(),
+      )
+      .max(MAX_AGENT_DYNAMIC_WORKFLOW_SUBAGENTS)
+      .optional()
+      .describe(
+        'Per-subagent entries with their own subagent_type, model, and thinking. Use instead of items when the subagents differ; tasks and items cannot be combined.',
+      ),
     model: z
       .string()
       .optional()
       .describe(
         'Which model to run the item-spawned subagents on: one of the aliases listed under "Available models" in this tool description, or "primary" for the main model you are running on (for hard, quality-sensitive tasks). When omitted, the configured default model is used. Resumed subagents always keep their own model.',
       ),
   })
-  .strict();
+  .strict()
+  .refine((input) => !(input.tasks !== undefined && input.items !== undefined), {
+    message: 'tasks and items cannot be combined; use one of them.',
+    path: ['tasks'],
+  });
 
 export type AgentDynamicWorkflowToolInput = z.infer<typeof AgentDynamicWorkflowToolInputSchema>;
 

packages/agent-core-v2/src/features/dynamic_workflow/tools/agent-dynamic_workflow/agentDynamicWorkflowTool.ts

@@ -48,6 +48,11 @@ interface AgentDynamicWorkflowSpawnSpec {
   readonly index: number;
   readonly item: string;
   readonly prompt: string;
+  readonly task?: {
+    readonly subagent_type?: string;
+    readonly model?: string;
+    readonly thinking?: string;
+  };
 }
 
 interface AgentDynamicWorkflowResumeSpec {
@@ -173,12 +178,41 @@ export class AgentDynamicWorkflowTool implements IAgentDynamicWorkflowTool {
         fork,
       });
     }
-    const profileName = plan?.profileName ?? DEFAULT_SUBAGENT_TYPE;
     const timeoutMs = resolveDynamicWorkflowTimeoutMs(this.config);
     const specs = await createAgentDynamicWorkflowSpecs(args, (agentId) =>
       this.dynamicWorkflowService.getDynamicWorkflowItem({ callerAgentId: this.callerAgentId, agentId }),
     );
+    const plansByIndex = new Map<number, SubagentSpawnPlan>();
+    for (const spec of specs) {
+      if (spec.kind !== 'spawn') continue;
+      if (spec.task === undefined) {
+        plansByIndex.set(spec.index, plan!);
+        continue;
+      }
+      const profileName = spec.task.subagent_type ?? args.defaults?.subagent_type ?? args.subagent_type;
+      if (fork) {
+        const incompatible = forkIncompatibility(
+          { subagent_type: profileName, model: spec.task.model ?? args.model },
+          this.profile.data(),
+        );
+        if (incompatible !== undefined) {
+          throw new Error2(ErrorCodes.VALIDATION_FAILED, incompatible);
+        }
+      }
+      plansByIndex.set(
+        spec.index,
+        await this.subagents.planSpawn({
+          callerAgentId: this.callerAgentId,
+          profileName,
+          model: spec.task.model ?? args.model,
+          thinking: spec.task.thinking,
+          fork,
+        }),
+      );
+    }
     const tasks: SessionDynamicWorkflowTask<AgentDynamicWorkflowSpec>[] = specs.map((spec) => {
+      const specPlan = spec.kind === 'spawn' ? plansByIndex.get(spec.index) : undefined;
+      const profileName = specPlan?.profileName ?? DEFAULT_SUBAGENT_TYPE;
       const descriptionName = spec.kind === 'resume' ? 'resume' : profileName;
       const common = {
         data: spec,
@@ -202,7 +236,7 @@ export class AgentDynamicWorkflowTool implements IAgentDynamicWorkflowTool {
       return {
         ...common,
         kind: 'spawn' as const,
-        plan: plan!,
+        plan: specPlan!,
       };
     });
     const results = await this.dynamicWorkflowService.run({
@@ -223,7 +257,8 @@ async function createAgentDynamicWorkflowSpecs(
     agentId: agentId.trim(),
     prompt: prompt.trim(),
   }));
-  const items = (args.items ?? []).map((item) => item.trim());
+  const taskEntries = (args.tasks ?? []).map((task) => ({ ...task, item: task.item.trim() }));
+  const items = taskEntries.length > 0 ? taskEntries.map((task) => task.item) : (args.items ?? []).map((item) => item.trim());
   const itemCount = items.length;
   const resumeCount = resumeEntries.length;
   const totalCount = resumeCount + itemCount;
@@ -279,11 +314,13 @@ async function createAgentDynamicWorkflowSpecs(
         );
       }
       seenPrompts.set(prompt, index + 1);
+      const task = taskEntries[index];
       specs.push({
         kind: 'spawn',
         index: specs.length + 1,
         item,
         prompt,
+        task: task === undefined ? undefined : { subagent_type: task.subagent_type, model: task.model, thinking: task.thinking },
       });
     });
   }

packages/agent-core-v2/src/session/subagent/spawn.ts

@@ -49,6 +49,7 @@ export interface SubagentSpawnPlanInput {
   readonly callerAgentId: string;
   readonly profileName?: string;
   readonly model?: string;
+  readonly thinking?: string;
   readonly fork?: boolean;
 }
 

packages/agent-core-v2/src/session/subagent/subagentRoutingService.ts

@@ -14,6 +14,8 @@ import { IAgentProfileService } from '#/agent/profile/profile';
 import { IConfigService } from '#/app/config/config';
 import { IModelCatalog, type Model } from '#/kosong/model/catalog';
 import { IAgentLifecycleService, MAIN_AGENT_ID } from '#/session/agentLifecycle/agentLifecycle';
+import type { SubagentSpawnPlanResolvedEvent } from '#/app/telemetry/events';
+import { ITelemetryService } from '#/app/telemetry/telemetry';
 
 import { IAgentBindingProvenanceService } from './bindingProvenance';
 import { resolveSubagentThinking, wrapSubagentModelError } from './configSection';
@@ -57,6 +59,7 @@ export class SessionSubagentRoutingService implements ISubagentRoutingService {
     @IConfigService private readonly configService: IConfigService,
     @IModelCatalog private readonly modelCatalog: IModelCatalog,
     @ISubagentModelPolicyService private readonly policy: ISubagentModelPolicyService,
+    @ITelemetryService private readonly telemetry: ITelemetryService,
   ) {}
 
   async resolve(input: SubagentSpawnPlanInput): Promise<ResolvedSpawnPlan> {
@@ -115,27 +118,44 @@ export class SessionSubagentRoutingService implements ISubagentRoutingService {
       throw wrapSubagentModelError(error, route.model, own.modelAlias);
     }
     const operation = fork ? 'fork' : 'spawn';
-    return {
+    const routing: SubagentBindingProvenance = {
+      operation,
+      profileSource,
+      modelSource: route.source,
+      policyMode: effective.effectivePolicy.mode,
+      policySource: effective.policySource,
+      featureSource: effective.feature.source,
+      resolvedFromRoutingEnvironmentRevision: environmentRevision,
+      routeDecisionFingerprint: routeDecisionFingerprint({
+        routingEnvironmentRevision: environmentRevision,
+        operation,
+        profile: requested,
+        model: input.model,
+        thinking: input.thinking,
+      }),
+    };
+    const plan: ResolvedSpawnPlan = {
       profileName: profile?.name ?? requestedProfileName,
       model: route.model,
-      thinking: resolveSubagentThinking(this.configService, model, route.thinking),
+      thinking: resolveSubagentThinking(this.configService, model, input.thinking ?? route.thinking),
       fork,
-      routing: {
-        operation,
-        profileSource,
-        modelSource: route.source,
-        policyMode: effective.effectivePolicy.mode,
-        policySource: effective.policySource,
-        featureSource: effective.feature.source,
-        resolvedFromRoutingEnvironmentRevision: environmentRevision,
-        routeDecisionFingerprint: routeDecisionFingerprint({
-          routingEnvironmentRevision: environmentRevision,
-          operation,
-          profile: requested,
-          model: input.model,
-        }),
-      },
+      routing,
+    };
+    const telemetryEvent: SubagentSpawnPlanResolvedEvent = {
+      operation,
+      profile_source: profileSource,
+      model_source: route.source,
+      policy_mode: routing.policyMode,
+      policy_source: routing.policySource,
+      feature_source: routing.featureSource,
+      routing_env_revision: environmentRevision,
+      route_decision: routing.routeDecisionFingerprint,
+      explicit_profile: requested !== undefined,
+      explicit_model: input.model !== undefined,
+      explicit_thinking: input.thinking !== undefined,
     };
+    this.telemetry.track2('subagent_spawn_plan_resolved', telemetryEvent);
+    return plan;
   }
 
   resumed(callerAgentId: string, child: IAgentScopeHandle): ResumedSubagentRouting {

packages/agent-core-v2/test/agent/fullCompaction/fullCompaction.test.ts

@@ -298,7 +298,7 @@ describe('FullCompaction', () => {
       properties: expect.objectContaining({
         agent_id: 'main',
         source: 'manual',
-        tokens_before: 6_257,
+        tokens_before: 6_504,
         tokens_after: expect.any(Number),
         duration_ms: expect.any(Number),
         compacted_count: 6,
@@ -572,7 +572,7 @@ describe('FullCompaction', () => {
       session_id: 'test-session',
       cwd: dir,
       trigger: 'auto',
-      token_count: 6_257,
+      token_count: 6_504,
     });
     expect(post).toMatchObject({
       hook_event_name: 'PostCompact',
@@ -658,7 +658,7 @@ describe('FullCompaction', () => {
       event: 'compaction_finished',
       properties: expect.objectContaining({
         source: 'manual',
-        tokens_before: 17_943,
+        tokens_before: 18_190,
         retry_count: 1,
         trace_id: 'trace-compact-1',
       }),
@@ -1125,7 +1125,7 @@ describe('FullCompaction', () => {
       properties: expect.objectContaining({
         agent_id: 'main',
         source: 'manual',
-        tokens_before: 17_943,
+        tokens_before: 18_190,
         duration_ms: expect.any(Number),
         round: 1,
         retry_count: 0,
@@ -1350,7 +1350,7 @@ describe('FullCompaction', () => {
       event: 'compaction_failed',
       properties: expect.objectContaining({
         source: 'manual',
-        tokens_before: 17_943,
+        tokens_before: 18_190,
         duration_ms: expect.any(Number),
         retry_count: 4,
         error_type: 'APIConnectionError',
@@ -1723,8 +1723,8 @@ describe('FullCompaction', () => {
       event: 'compaction_finished',
       properties: expect.objectContaining({
         source: 'auto',
-        tokens_before: 6_264,
-        tokens_after: 6_248,
+        tokens_before: 6_511,
+        tokens_after: 6_495,
         compacted_count: 7,
         retry_count: 0,
       }),

packages/agent-core-v2/test/agent/loop/loop.test.ts

@@ -1,3 +1,4 @@
+import { ITelemetryService } from '#/app/telemetry/telemetry';
 import { SubagentModelPolicyService } from '#/session/subagent/subagentModelPolicyService';
 import { SessionSubagentRoutingService } from '#/session/subagent/subagentRoutingService';
 import { afterEach, beforeEach, describe, expect, it, vi, type Mock } from 'vitest';
@@ -275,6 +276,7 @@ function realSubagents(
     config,
     modelCatalog,
     new SubagentModelPolicyService(config, flags, modelCatalog),
+    { _serviceBrand: undefined, track2: vi.fn(), track: vi.fn() } as unknown as ITelemetryService,
   );
   return new SessionSubagentService(lifecycle, catalog, sessionContext, stubLog(), routing);
 }
@@ -665,6 +667,50 @@ describe('dynamic_workflow context reconciliation', () => {
 });
 
 describe('AgentDynamicWorkflowTool', () => {
+  it('resolves a plan per task so a workflow can mix subagent types, models, and thinking', async () => {
+    const host = mockDynamicWorkflowHost({
+      run: vi.fn().mockImplementation(async ({ tasks }) =>
+        tasks.map((task: { kind: string; data: { index: number; item?: string }; plan?: unknown }) => ({
+          task,
+          agentId: `agent-${task.data.index}`,
+          status: 'completed',
+          result: `done ${task.data.item ?? ''}`,
+        })),
+      ),
+    });
+    const dynamicWorkflowMode = mockDynamicWorkflowMode();
+    const cfg = stubConfig({
+      defaultModel: 'provider/fast',
+      models: { 'provider/fast': 'fast and cheap', 'provider/smart': 'hard tasks' },
+    });
+    const tool = new AgentDynamicWorkflowTool(host.dynamicWorkflowService, makeAgentScopeContext({ agentId: host.callerAgentId, agentScope: '' }), dynamicWorkflowMode, cfg, stubFlag(true), realSubagents(stubDynamicWorkflowCatalog(), cfg, stubFlag(true), stubCallerProfile()), stubCallerProfile());
+    const input = {
+      description: 'Review files',
+      prompt_template: 'Review {{item}}',
+      defaults: { subagent_type: 'explore' },
+      tasks: [
+        { item: 'src/a.ts' },
+        { item: 'src/b.ts', subagent_type: 'coder', model: 'provider/smart', thinking: 'low' },
+        { item: 'src/c.ts', model: 'primary' },
+      ],
+    };
+    expect(AgentDynamicWorkflowToolInputSchema.safeParse(input).success).toBe(true);
+    expect(AgentDynamicWorkflowToolInputSchema.safeParse({ ...input, items: ['src/d.ts'] }).success).toBe(false);
+
+    const result = await executeTool(tool, context(input));
+    expect(result.isError).toBeUndefined();
+    const call = (host.dynamicWorkflowService.run as ReturnType<typeof vi.fn>).mock.calls[0]![0] as {
+      tasks: Array<{ kind: string; profileName: string; plan: { profileName: string; model: string; thinking?: string; routing?: { modelSource: string } } }>;
+    };
+    expect(call.tasks.map((task) => [task.profileName, task.plan.model, task.plan.thinking, task.plan.routing?.modelSource])).toEqual([
+      ['explore', 'provider/fast', undefined, 'policy-pool'],
+      ['coder', 'provider/smart', 'low', 'policy-pool'],
+      ['explore', 'mock-model', 'off', 'caller'],
+    ]);
+    expect(result.output).toContain('<subagent agent_id="agent-1" item="src/a.ts"');
+    expect(result.output).toContain('<subagent agent_id="agent-2" item="src/b.ts"');
+  });
+
   it('renders durable binding attributes on each subagent row and escapes them', async () => {
     const host = mockDynamicWorkflowHost({
       run: vi.fn().mockResolvedValue([