Skip to content

Commit 8f2b5c6

Browse files
dvyukovdvyukov
committed
syz-agent: add agentic server
Add server for running agentic workflows as part of syzbot. The architecture and use are similar to that of syz-ci.
1 parent 44dae31 commit 8f2b5c6

File tree

3 files changed

+208
-2
lines changed

3 files changed

+208
-2
lines changed

‎Makefile‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -104,7 +104,7 @@ ifeq ("$(TARGETOS)", "trusty")
104104
endif
105105

106106
.PHONY: all clean host target \
107-
manager executor kfuzztest ci hub \
107+
manager executor kfuzztest ci hub agent \
108108
execprog mutate prog2c trace2syz repro upgrade db \
109109
usbgen symbolize cover kconf syz-build crush \
110110
bin/syz-extract bin/syz-fmt \
@@ -172,6 +172,9 @@ ci: descriptions
172172
hub: descriptions
173173
GOOS=$(HOSTOS) GOARCH=$(HOSTARCH) $(HOSTGO) build $(GOHOSTFLAGS) -o ./bin/syz-hub github.com/google/syzkaller/syz-hub
174174

175+
agent: descriptions
176+
GOOS=$(HOSTOS) GOARCH=$(HOSTARCH) $(HOSTGO) build $(GOHOSTFLAGS) -o ./bin/syz-agent github.com/google/syzkaller/syz-agent
177+
175178
repro: descriptions
176179
GOOS=$(HOSTOS) GOARCH=$(HOSTARCH) $(HOSTGO) build $(GOHOSTFLAGS) -o ./bin/syz-repro github.com/google/syzkaller/tools/syz-repro
177180

‎pkg/updater/updater.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -274,7 +274,7 @@ func (upd *Updater) build(commit *vcs.Commit) error {
274274
}
275275
}
276276
// This will also generate descriptions and should go before the 'go test' below.
277-
cmd := osutil.Command(instance.MakeBin, "host", "ci")
277+
cmd := osutil.Command(instance.MakeBin, "host", "ci", "agent")
278278
cmd.Dir = upd.syzkallerDir
279279
cmd.Env = append([]string{"GOPATH=" + upd.gopathDir}, os.Environ()...)
280280
if _, err := osutil.Run(time.Hour, cmd); err != nil {

‎syz-agent/agent.go‎

Lines changed: 203 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,203 @@
1+
// Copyright 2025 syzkaller project authors. All rights reserved.
2+
// Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.
3+
4+
package main
5+
6+
import (
7+
"context"
8+
"encoding/json"
9+
"flag"
10+
"fmt"
11+
"maps"
12+
"path/filepath"
13+
"slices"
14+
"strings"
15+
"sync"
16+
"time"
17+
18+
"github.com/google/syzkaller/dashboard/dashapi"
19+
"github.com/google/syzkaller/pkg/aflow"
20+
"github.com/google/syzkaller/pkg/aflow/flow/patching"
21+
"github.com/google/syzkaller/pkg/aflow/journal"
22+
"github.com/google/syzkaller/pkg/config"
23+
"github.com/google/syzkaller/pkg/log"
24+
"github.com/google/syzkaller/pkg/mgrconfig"
25+
"github.com/google/syzkaller/pkg/osutil"
26+
"github.com/google/syzkaller/pkg/tool"
27+
"github.com/google/syzkaller/pkg/updater"
28+
)
29+
30+
type Config struct {
31+
DashboardAddr string `json:"dashboard_addr"`
32+
DashboardClient string `json:"dashboard_client"` // Global non-namespace client.
33+
DashboardKey string `json:"dashboard_key"`
34+
SyzkallerRepo string `json:"syzkaller_repo"`
35+
SyzkallerBranch string `json:"syzkaller_branch"`
36+
CodesearchToolBin string `json:"codesearch_tool_bin"`
37+
KernelConfig string `json:"kernel_config"`
38+
Target string `json:"target"`
39+
Image string `json:"image"`
40+
Type string `json:"type"`
41+
VM json.RawMessage `json:"vm"`
42+
}
43+
44+
func main() {
45+
var (
46+
flagConfig = flag.String("config", "", "config file")
47+
flagExitOnUpgrade = flag.Bool("exit-on-upgrade", false,
48+
"exit after a syz-ci upgrade is applied; otherwise syz-ci restarts")
49+
flagAutoUpdate = flag.Bool("autoupdate", true, "auto-update the binary (for testing)")
50+
)
51+
defer tool.Init()()
52+
log.SetName("syz-agent")
53+
if err := run(*flagConfig, *flagExitOnUpgrade, *flagAutoUpdate); err != nil {
54+
log.Fatal(err)
55+
}
56+
}
57+
58+
func run(configFile string, exitOnUpgrade, autoUpdate bool) error {
59+
cfg := &Config{
60+
SyzkallerRepo: "https://github.com/google/syzkaller.git",
61+
SyzkallerBranch: "master",
62+
}
63+
if err := config.LoadFile(configFile, cfg); err != nil {
64+
return fmt.Errorf("failed to load config: %w", err)
65+
}
66+
os, vmarch, arch, _, _, err := mgrconfig.SplitTarget(cfg.Target)
67+
if err != nil {
68+
return err
69+
}
70+
dash, err := dashapi.New(cfg.DashboardClient, cfg.DashboardAddr, cfg.DashboardKey)
71+
if err != nil {
72+
return err
73+
}
74+
buildSem := osutil.NewSemaphore(1)
75+
updater, err := updater.New(&updater.Config{
76+
ExitOnUpdate: exitOnUpgrade,
77+
BuildSem: buildSem,
78+
SyzkallerRepo: cfg.SyzkallerRepo,
79+
SyzkallerBranch: cfg.SyzkallerBranch,
80+
Targets: map[updater.Target]bool{
81+
updater.Target{
82+
OS: os,
83+
VMArch: vmarch,
84+
Arch: arch,
85+
}: true,
86+
},
87+
})
88+
if err != nil {
89+
return err
90+
}
91+
updatePending := make(chan struct{})
92+
shutdownPending := make(chan struct{})
93+
osutil.HandleInterrupts(shutdownPending)
94+
updater.UpdateOnStart(autoUpdate, updatePending, shutdownPending)
95+
96+
ctx, stop := context.WithCancel(context.Background())
97+
var wg sync.WaitGroup
98+
wg.Add(1)
99+
go func() {
100+
defer wg.Done()
101+
for {
102+
ok, err := poll(ctx, cfg, dash)
103+
if err != nil {
104+
log.Error(err)
105+
}
106+
var delay time.Duration
107+
if !ok {
108+
// Don't poll dashboard too often, if there are no jobs,
109+
// or errors are happenning.
110+
delay = 10 * time.Second
111+
}
112+
select {
113+
case <-ctx.Done():
114+
return
115+
case <-time.After(delay):
116+
}
117+
}
118+
}()
119+
120+
select {
121+
case <-shutdownPending:
122+
case <-updatePending:
123+
}
124+
stop()
125+
wg.Wait()
126+
127+
select {
128+
case <-shutdownPending:
129+
default:
130+
updater.UpdateAndRestart()
131+
}
132+
return nil
133+
}
134+
135+
func poll(ctx context.Context, cfg *Config, dash *dashapi.Dashboard) (bool, error) {
136+
req := &dashapi.AIJobPollReq{
137+
Workflows: slices.Collect(maps.Keys(aflow.Flows)),
138+
}
139+
resp, err := dash.AIJobPoll(req)
140+
if err != nil {
141+
return false, err
142+
}
143+
if resp.ID == "" {
144+
return false, nil
145+
}
146+
flow := aflow.Flows[resp.Workflow]
147+
if flow == nil {
148+
return false, fmt.Errorf("unsupported flow %q", resp.Workflow)
149+
}
150+
doneReq := &dashapi.AIJobDoneReq{
151+
ID: resp.ID,
152+
}
153+
switch flow.Type {
154+
case patching.Type:
155+
doneReq.Patching, err = patchingJob(ctx, cfg, dash, flow, resp)
156+
default:
157+
err = fmt.Errorf("unsupported flow type %q", flow.Type)
158+
}
159+
if err != nil {
160+
doneReq.Error = err.Error()
161+
}
162+
if err := dash.AIJobDone(doneReq); err != nil {
163+
return false, err
164+
}
165+
return true, nil
166+
}
167+
168+
func patchingJob(ctx context.Context, cfg *Config, dash *dashapi.Dashboard, flow *aflow.Flow,
169+
req *dashapi.AIJobPollResp) (*dashapi.AIPatchingResult, error) {
170+
inputs := patching.Inputs{
171+
ReproOpts: req.Patching.ReproOpts,
172+
ReproSyz: req.Patching.ReproSyz,
173+
ReproC: req.Patching.ReproC,
174+
KernelConfig: req.Patching.KernelConfig,
175+
SyzkallerCommit: req.Patching.SyzkallerCommit,
176+
CodesearchToolBin: cfg.CodesearchToolBin,
177+
Syzkaller: osutil.Abs(filepath.FromSlash("syzkaller/current")),
178+
Image: cfg.Image,
179+
Type: cfg.Type,
180+
VM: cfg.VM,
181+
}
182+
onEvent := func(ev *journal.Event) error {
183+
log.Logf(0, "%v%v", strings.Repeat(" ", ev.Nesting), ev.Description())
184+
dump, err := json.MarshalIndent(ev, "", "\t")
185+
if err != nil {
186+
panic(err)
187+
}
188+
fmt.Printf("%s\n\n", dump)
189+
return dash.AIJournal(&dashapi.AIJournalReq{
190+
JobID: req.ID,
191+
Event: ev,
192+
})
193+
}
194+
res, err := flow.Execute(ctx, false, osutil.Abs("workdir"), inputs, nil, onEvent)
195+
if err != nil {
196+
return nil, err
197+
}
198+
outputs := res.(patching.Outputs)
199+
return &dashapi.AIPatchingResult{
200+
PatchDescription: outputs.PatchDescription,
201+
PatchDiff: outputs.PatchDiff,
202+
}, nil
203+
}

0 commit comments

Comments
 (0)