Skip to content

feat(workflow): run workflows on ADK's graph engine #136

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Cidan merged 1 commit into from
Aug 21, 2026
Merged

feat(workflow): run workflows on ADK's graph engine #136

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
221 changes: 100 additions & 121 deletions CLAUDE.md
View file
Open in desktop

Large diffs are not rendered by default.

34 changes: 24 additions & 10 deletions cmd/ask/agent_run.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,8 @@ import (
tea "charm.land/bubbletea/v2"
"github.com/Cidan/ask/pkg/engine"
"github.com/Cidan/ask/pkg/tools"
"google.golang.org/adk/v2/agent"
"github.com/Cidan/ask/pkg/workflow"
adkagent "google.golang.org/adk/v2/agent"
"google.golang.org/adk/v2/agent/llmagent"
adkmodel "google.golang.org/adk/v2/model"
"google.golang.org/adk/v2/session"
Expand Down Expand Up @@ -72,6 +73,13 @@ type agentSession struct {
retryMaxRetries int
retryInitialDelay time.Duration
retryBackoffFactor float64

// workflowAgent, when set, replaces the ask_coder agent for this
// session's turns: the session runs a compiled workflow graph
// instead of a single coder agent. workflowProgress consumes the
// same ADK event stream to drive the workflow tab's step log.
workflowAgent adkagent.Agent
workflowProgress *workflow.Progress
}

func (s *agentSession) refreshToolset() {
Expand Down Expand Up @@ -356,14 +364,19 @@ func (s *agentSession) runTurn(turn agentTurn) {
toolsets = append(toolsets, skillTS)
}

agentInstance, err := llmagent.New(llmagent.Config{
Name: "ask_coder",
Model: llm,
InstructionProvider: instructionProvider,
Tools: adkTools,
Toolsets: toolsets,
GenerateContentConfig: genaiConfig,
})
var agentInstance adkagent.Agent
if s.workflowAgent != nil {
agentInstance = s.workflowAgent
} else {
agentInstance, err = llmagent.New(llmagent.Config{
Name: "ask_coder",
Model: llm,
InstructionProvider: instructionProvider,
Tools: adkTools,
Toolsets: toolsets,
GenerateContentConfig: genaiConfig,
})
}
if err != nil {
s.emit(providerDoneMsg{
res: providerResult{SessionID: s.sessionID, IsError: true, Result: err.Error()},
Expand Down Expand Up @@ -404,7 +417,7 @@ func (s *agentSession) runTurn(turn agentTurn) {
displayNames := make(map[string]string)
backgroundCalls := make(map[string]bool)

for event, err := range r.Run(ctx, "user", s.sessionID, adkUserMsg, agent.RunConfig{}) {
for event, err := range r.Run(ctx, "user", s.sessionID, adkUserMsg, adkagent.RunConfig{}) {
if err != nil {
if isAgentCancel(err) {
s.emit(providerDoneMsg{res: providerResult{SessionID: s.sessionID}})
Expand All @@ -421,6 +434,7 @@ func (s *agentSession) runTurn(turn agentTurn) {
if event == nil {
continue
}
s.workflowProgress.Observe(event)

if event.UsageMetadata != nil {
usage := TokenUsage{
Expand Down
28 changes: 3 additions & 25 deletions cmd/ask/aliases.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,10 +84,6 @@ func currentWorkflowStepMeta(r *workflowRunState) (name, provider, model string)
return "", "", ""
}
top := r.Workflow.Steps[r.StepIdx]
if r.loop != nil && top.IsLoop() && r.loop.innerIdx < len(top.Steps) {
inner := top.Steps[r.loop.innerIdx]
return inner.Name, inner.Provider, inner.Model
}
return top.Name, top.Provider, top.Model
}

Expand Down Expand Up @@ -116,20 +112,12 @@ func toPkgWorkflowStep(s workflowStep) workflow.Step {
}
}

func buildWorkflowStepPrompt(step workflowStep, source workflowSource, prevOutputs []string, pc *stepPromptCtx) string {
return workflow.BuildStepPrompt(toPkgWorkflowStep(step), source, prevOutputs, pc)
func buildWorkflowStepInstruction(step workflowStep, source workflowSource, pc *stepPromptCtx) string {
return workflow.BuildStepInstruction(toPkgWorkflowStep(step), source, pc)
}

var (
stepNotesDir = workflow.StepNotesDir
startPlanDir = workflow.StartPlanDir
ensureStartPlanExists = workflow.EnsureStartPlanExists
ensureStepNotesDir = workflow.EnsureStepNotesDir
removeAllWorkflowPlans = workflow.RemoveAllWorkflowPlans
clearWorkflowPlans = workflow.ClearWorkflowPlans
workflowPlansDir = workflow.PlansDir
loopNoteLine = workflow.LoopNoteLine
sanitizeStepName = workflow.SanitizeStepName
loopNoteLine = workflow.LoopNoteLine
)

func lastOf(s []string) string {
Expand Down Expand Up @@ -167,7 +155,6 @@ var (
unwrapInvokeToolCall = tools.UnwrapInvokeToolCall
runAskPassHelper = tools.RunAskPassHelper
applyBashFilter = tools.ApplyBashFilter
clearPlansCore = tools.ClearPlansCore
agentAskUserQuestionTool = tools.AskUserQuestionTool
agentEndTurnTool = tools.EndTurnTool
agentFinalizedPlanTool = tools.FinalizedPlanTool
Expand All @@ -191,15 +178,6 @@ var (
agentPreloadMemoryTool = tools.PreloadMemoryTool
)

const (
clearPlansToolDescription = tools.ClearPlansToolDescription
)

type (
clearPlansInput = tools.ClearPlansInput
clearPlansOutput = tools.ClearPlansOutput
)

func errResult(text string) *mcp.CallToolResult {
return &mcp.CallToolResult{
IsError: true,
Expand Down
53 changes: 19 additions & 34 deletions cmd/ask/chat_workflow_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -129,7 +129,7 @@ func TestWorkflowSource_RefBlock_ChatFormat(t *testing.T) {

// TestWorkflowSource_RefBlock_EmptyChatReturnsEmpty guards the
// "skip the section entirely" path — an empty transcript must
// not emit a dangling header. buildWorkflowStepPrompt relies on
// not emit a dangling header. buildWorkflowStepInstruction relies on
// this to drop the reference block when there's nothing to
// reference.
func TestWorkflowSource_RefBlock_EmptyChatReturnsEmpty(t *testing.T) {
Expand All @@ -151,11 +151,10 @@ func TestWorkflowSource_RefBlock_IssueFormat(t *testing.T) {
}
}

// TestBuildWorkflowStepPrompt_ChatSource verifies the prompt
// assembly for a chat-sourced workflow. Step 0 should carry the
// transcript reference; a later step should layer the previous-step
// output block under the same reference.
func TestBuildWorkflowStepPrompt_ChatSource(t *testing.T) {
// TestBuildWorkflowStepInstruction_ChatSource verifies instruction
// assembly for a chat-sourced workflow: the transcript reference rides
// along with the author's prompt and the end_turn contract.
func TestBuildWorkflowStepInstruction_ChatSource(t *testing.T) {
step := workflowStep{Prompt: "Summarise."}
source := workflowSource{
Kind: workflowSourceChat,
Expand All @@ -165,36 +164,22 @@ func TestBuildWorkflowStepPrompt_ChatSource(t *testing.T) {
},
}

step0 := buildWorkflowStepPrompt(step, source, nil, nil)
if !strings.Contains(step0, "Summarise.") {
t.Errorf("step 0 must include user prompt; got %q", step0)
}
if !strings.Contains(step0, "Reference (chat transcript):") {
t.Errorf("step 0 must include chat transcript header; got %q", step0)
}
if !strings.Contains(step0, "user: what's a goroutine?") {
t.Errorf("step 0 must include user turn; got %q", step0)
}
if !strings.Contains(step0, "assistant: a green-thread primitive.") {
t.Errorf("step 0 must include assistant turn; got %q", step0)
}
if strings.Contains(step0, "Previous step output:") {
t.Errorf("step 0 must NOT include previous-step block; got %q", step0)
}
if strings.Contains(step0, "Reference: ") {
// Make sure we didn't accidentally emit the issue-style line.
t.Errorf("chat source must NOT emit issue-style Reference line; got %q", step0)
}

stepN := buildWorkflowStepPrompt(step, source, []string{"prior step output text"}, nil)
if !strings.Contains(stepN, "Previous step output:") {
t.Errorf("step N must include previous-step block; got %q", stepN)
got := buildWorkflowStepInstruction(step, source, nil)
for _, want := range []string{
"Summarise.",
"Reference (chat transcript):",
"user: what's a goroutine?",
"assistant: a green-thread primitive.",
} {
if !strings.Contains(got, want) {
t.Errorf("instruction missing %q; got %q", want, got)
}
}
if !strings.Contains(stepN, "prior step output text") {
t.Errorf("step N must include the log entry; got %q", stepN)
if strings.Contains(got, "Previous step output:") {
t.Errorf("previous-step threading is the graph's job now; got %q", got)
}
if !strings.Contains(stepN, "Reference (chat transcript):") {
t.Errorf("step N must still include chat transcript; got %q", stepN)
if strings.Contains(got, "Reference: ") {
t.Errorf("chat source must not emit the issue-style Reference line; got %q", got)
}
}

Expand Down
101 changes: 3 additions & 98 deletions cmd/ask/coordinator.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,6 @@ package main

import (
"context"
"errors"
"fmt"
"strings"
"sync"

tea "charm.land/bubbletea/v2"
Expand Down Expand Up @@ -280,99 +277,8 @@ func (l tuiWorkflowListener) OnNote(tabID int, text string) {
})
}

// ExecuteStep implements workflow.StepExecutor for Coordinator.
func (c *Coordinator) ExecuteStep(ctx context.Context, cwd string, tabID int, step workflow.Step, prompt string, isFinal bool) (workflow.StepResult, error) {
prov := providerByID(step.Provider)
if prov == nil {
return workflow.StepResult{}, fmt.Errorf("provider not registered: %s", step.Provider)
}

args := ProviderSessionArgs{
Cwd: cwd,
TabID: tabID,
Model: step.Model,
Effort: "medium",
SkipAllPermissions: true,
InWorkflow: true,
IsWorkflowFinalStep: isFinal,
}

proc, ch, err := prov.StartSession(args)
if err != nil {
return workflow.StepResult{}, err
}

session, ok := proc.payload.(*agentSession)
if !ok {
return workflow.StepResult{}, errors.New("proc payload is not an agent session")
}
c.SetSession(tabID, session)

err = session.queueTurn(prompt)
if err != nil {
session.shutdown()
c.RemoveSession(tabID)
return workflow.StepResult{}, err
}

var stepResult string
var stepErr error
stepLoop:
for msg := range ch {
switch m := msg.(type) {
case assistantTextMsg:
stepResult += m.text
case providerDoneMsg:
if m.err != nil {
stepErr = m.err
} else if m.res.IsError {
stepErr = fmt.Errorf("step failed: %s", m.res.Result)
} else {
stepResult = m.res.Result
}
case turnCompleteMsg:
break stepLoop
}
}

session.shutdown()
c.RemoveSession(tabID)

if stepErr != nil {
return workflow.StepResult{}, stepErr
}

summary := ""
decision := ""
if session.env.PendingEndTurn != nil {
summary = session.env.PendingEndTurn.Summary
decision = session.env.PendingEndTurn.Decision
}
if summary == "" && strings.TrimSpace(stepResult) != "" {
firstLine := strings.TrimSpace(strings.Split(strings.TrimSpace(stepResult), "\n")[0])
if len(firstLine) > 200 {
firstLine = firstLine[:200] + "…"
}
summary = firstLine
}

var finishData *workflow.FinishData
if session.env.PendingFinishData != nil {
finishData = &workflow.FinishData{
Description: session.env.PendingFinishData.Description,
Artifacts: session.env.PendingFinishData.Artifacts,
}
}

return workflow.StepResult{
Output: stepResult,
Summary: summary,
Decision: decision,
FinishData: finishData,
}, nil
}

// RunWorkflow executes a workflow synchronously step by step in the background.
// RunWorkflow compiles the definition to an ADK workflow graph and drives
// it to completion in the background.
func (c *Coordinator) RunWorkflow(ctx context.Context, tabID int, def workflowDef, src workflowSource) (finalizedPlanReply, error) {
ctx, cancel := context.WithCancel(ctx)
defer cancel()
Expand Down Expand Up @@ -407,8 +313,7 @@ func (c *Coordinator) RunWorkflow(ctx context.Context, tabID int, def workflowDe
}

listener := tuiWorkflowListener{tabID: tabID}
runner := workflow.NewRunner(workflow.GlobalTracker(), c, listener)
runState, err := runner.Run(ctx, rootCwd, tabID, toPkgWorkflowDef(def), src)
runState, err := c.runWorkflowGraph(ctx, rootCwd, tabID, toPkgWorkflowDef(def), src, listener)
if err != nil {
return finalizedPlanReply{}, err
}
Expand Down
Loading