{"record":{"id":"4d5a60ec8ec6c808","repo":"shareAI-lab/learn-claude-code","slug":"workflow-nesting-is-one-level-only","errorCode":null,"errorMessage":"workflow() nesting is one level only","messagePattern":"workflow\\(\\) nesting is one level only","errorType":"validation","errorClass":"WorkflowInputError","httpStatus":null,"severity":"error","filePath":"s16_workflow_runtime/code.py","lineNumber":525,"sourceCode":"        \"\"\"BARRIER: run all thunks concurrently and fail if any thunk fails.\"\"\"\n        return await asyncio.gather(*[thunk() for thunk in thunks])\n\n    async def pipeline(self, items, *stages):\n        \"\"\"Per-item staged flow, NO barrier between stages: item A can be in\n        stage 3 while item B is still in stage 1. Each stage gets\n        (prev_result, original_item, index). A throwing stage fails the workflow.\"\"\"\n        async def run_item(item, idx):\n            value = item\n            for stage in stages:\n                value = await stage(value, item, idx)\n            return value\n        return await asyncio.gather(*[run_item(it, i) for i, it in enumerate(items)])\n\n    async def workflow(self, name, args=None):\n        \"\"\"Run a saved workflow inline as a child (one level), sharing this run's\n        journal + budget + agent counter.\"\"\"\n        if self._depth >= 1:\n            raise WorkflowInputError(\"workflow() nesting is one level only\")\n        if name not in WORKFLOWS:\n            raise WorkflowInputError(f\"unknown workflow '{name}'\")\n        meta, fn = WORKFLOWS[name]\n        child = ExecutionState(self.task, self.journal, self.runner, self.budget,\n                               args or {}, depth=self._depth + 1,\n                               limits=self._limits)\n        return await fn(child, args or {})\n\n\n# -- Workflow Tool --\nclass WorkflowTool:\n    \"\"\"The Workflow tool. .call() validates meta, runs the permission check,\n    creates runId/taskId, registers a LocalWorkflowTask, and emits lifecycle\n    events while executing the script. It returns the result and task state and\n    supports resume.\"\"\"\n\n    async def call(self, meta, script_fn, args=None, resume_from_run_id=None):\n        validate_meta(meta)","sourceCodeStart":507,"sourceCodeEnd":543,"githubUrl":"https://github.com/shareAI-lab/learn-claude-code/blob/985456f4adea6f4df8fbad4112245dbd97444eae/s16_workflow_runtime/code.py#L507-L543","documentation":"ExecutionState.workflow() runs a registered workflow inline as a child of the current run, sharing journal/budget/limits, but only one level of nesting is allowed. The child ExecutionState is created with depth=self._depth + 1, and any workflow() call at depth >= 1 raises WorkflowInputError('workflow() nesting is one level only'). This keeps the run graph flat and budgets/locks comprehensible.","triggerScenarios":"A workflow function calls ctx.workflow('child', ...) when it is itself already running as a child of another workflow — i.e. a grandchild composition attempt.","commonSituations":"Refactoring shared logic into composable workflows and chaining three deep; copy-pasting a workflow that itself embeds another workflow.","solutions":["Flatten the composition: inline the grandchild's logic (or extract it into plain helper functions) so no workflow() call happens inside a child.","Run the two workflows sequentially as separate top-level runs, feeding the first run's result into the second's args.","Move shared steps into a helper the parent calls via ctx.agent()/pipeline instead of a nested workflow()."],"exampleFix":"# before (inside child workflow)\nasync def child(ctx, args):\n    return await ctx.workflow(\"grandchild\", args)  # depth 1 -> raises\n\n# after\nasync def child(ctx, args):\n    # inline the grandchild steps directly at this level\n    out = await ctx.agent(..., label=\"grandchild:step\")\n    return out","handlingStrategy":"validation","validationCode":"def plan_nesting(workflow_fns, name) -> int:\n    # walk which workflows each fn calls via ctx.workflow(); reject depth > 1\n    depth = 0\n    for called in workflow_fns[name].calls_workflow():\n        depth = max(depth, 1 + plan_nesting(workflow_fns, called))\n    return depth\nassert plan_nesting(workflow_fns, entry) <= 1","typeGuard":null,"tryCatchPattern":"try:\n    await ctx.workflow(\"child\", args)\nexcept WorkflowInputError as e:\n    if \"nesting\" in str(e):\n        # flatten: run child as a separate top-level run\n        await run_workflow(\"child\", args)\n    else:\n        raise","preventionTips":["Design workflows leaf-first: helpers are plain functions, only top-level entry points are registered.","Enforce the one-level rule in review: grep for ctx.workflow( inside any workflow body other than entry points.","Chain workflows by passing runIds/results between top-level runs, not by nesting."],"tags":["workflow","nesting","composition","limits"],"backgroundTag":null,"analyzedSha":"985456f4adea6f4df8fbad4112245dbd97444eae","analyzedAt":"2026-08-14T22:02:26.028Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}