Skip to content

Commit 0be944a

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 d4b57bc commit 0be944a

File tree

3 files changed

+229
-2
lines changed

3 files changed

+229
-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: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,7 @@ func New(cfg *Config) (*Updater, error) {
8888
"tag": true, // contains syzkaller repo git hash
8989
"bin/syz-ci": true, // these are just copied from syzkaller dir
9090
"bin/syz-manager": true,
91+
"bin/syz-agent": true,
9192
"sys/*/test/*": true,
9293
}
9394
for target := range cfg.Targets {
@@ -273,7 +274,7 @@ func (upd *Updater) build(commit *vcs.Commit) error {
273274
}
274275
}
275276
// This will also generate descriptions and should go before the 'go test' below.
276-
cmd := osutil.Command(instance.MakeBin, "host", "ci")
277+
cmd := osutil.Command(instance.MakeBin, "host", "ci", "agent")
277278
cmd.Dir = upd.syzkallerDir
278279
cmd.Env = append([]string{"GOPATH=" + upd.gopathDir}, os.Environ()...)
279280
if _, err := osutil.Run(time.Hour, cmd); err != nil {

‎syz-agent/agent.go‎

Lines changed: 223 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,223 @@
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+
_ "net/http/pprof"
13+
"path/filepath"
14+
"sync"
15+
"time"
16+
17+
"github.com/google/syzkaller/dashboard/dashapi"
18+
"github.com/google/syzkaller/pkg/aflow"
19+
_ "github.com/google/syzkaller/pkg/aflow/flow"
20+
"github.com/google/syzkaller/pkg/aflow/trajectory"
21+
"github.com/google/syzkaller/pkg/config"
22+
"github.com/google/syzkaller/pkg/log"
23+
"github.com/google/syzkaller/pkg/mgrconfig"
24+
"github.com/google/syzkaller/pkg/osutil"
25+
"github.com/google/syzkaller/pkg/tool"
26+
"github.com/google/syzkaller/pkg/updater"
27+
"github.com/google/syzkaller/prog"
28+
)
29+
30+
type Config struct {
31+
// Currently serves only net/http/pprof handlers.
32+
HTTP string `json:"http"`
33+
DashboardAddr string `json:"dashboard_addr"`
34+
DashboardClient string `json:"dashboard_client"` // Global non-namespace client.
35+
DashboardKey string `json:"dashboard_key"`
36+
SyzkallerRepo string `json:"syzkaller_repo"`
37+
SyzkallerBranch string `json:"syzkaller_branch"`
38+
CodesearchToolBin string `json:"codesearch_tool_bin"`
39+
KernelConfig string `json:"kernel_config"`
40+
Target string `json:"target"`
41+
Image string `json:"image"`
42+
Type string `json:"type"`
43+
VM json.RawMessage `json:"vm"`
44+
// Use fixed base commit for patching jobs (for testing).
45+
FixedBaseCommit string `json:"fixed_base_commit"`
46+
// Use this LLM model (for testing, if empty use a default model).
47+
Model string `json:"model"`
48+
}
49+
50+
func main() {
51+
var (
52+
flagConfig = flag.String("config", "", "config file")
53+
flagExitOnUpgrade = flag.Bool("exit-on-upgrade", false,
54+
"exit after a syz-ci upgrade is applied; otherwise syz-ci restarts")
55+
flagAutoUpdate = flag.Bool("autoupdate", true, "auto-update the binary (for testing)")
56+
)
57+
defer tool.Init()()
58+
log.SetName("syz-agent")
59+
if err := run(*flagConfig, *flagExitOnUpgrade, *flagAutoUpdate); err != nil {
60+
log.Fatal(err)
61+
}
62+
}
63+
64+
func run(configFile string, exitOnUpgrade, autoUpdate bool) error {
65+
cfg := &Config{
66+
SyzkallerRepo: "https://github.com/google/syzkaller.git",
67+
SyzkallerBranch: "master",
68+
Model: aflow.DefaultModel,
69+
}
70+
if err := config.LoadFile(configFile, cfg); err != nil {
71+
return fmt.Errorf("failed to load config: %w", err)
72+
}
73+
tool.ServeHTTP(cfg.HTTP)
74+
os, vmarch, arch, _, _, err := mgrconfig.SplitTarget(cfg.Target)
75+
if err != nil {
76+
return err
77+
}
78+
dash, err := dashapi.New(cfg.DashboardClient, cfg.DashboardAddr, cfg.DashboardKey)
79+
if err != nil {
80+
return err
81+
}
82+
buildSem := osutil.NewSemaphore(1)
83+
updater, err := updater.New(&updater.Config{
84+
ExitOnUpdate: exitOnUpgrade,
85+
BuildSem: buildSem,
86+
SyzkallerRepo: cfg.SyzkallerRepo,
87+
SyzkallerBranch: cfg.SyzkallerBranch,
88+
Targets: map[updater.Target]bool{
89+
{
90+
OS: os,
91+
VMArch: vmarch,
92+
Arch: arch,
93+
}: true,
94+
},
95+
})
96+
if err != nil {
97+
return err
98+
}
99+
updatePending := make(chan struct{})
100+
shutdownPending := make(chan struct{})
101+
osutil.HandleInterrupts(shutdownPending)
102+
updater.UpdateOnStart(autoUpdate, updatePending, shutdownPending)
103+
104+
const workdir = "workdir"
105+
const cacheSize = 1 << 40 // 1TB should be enough for everyone!
106+
cache, err := aflow.NewCache(filepath.Join(workdir, "cache"), cacheSize)
107+
if err != nil {
108+
return err
109+
}
110+
111+
s := &Server{
112+
cfg: cfg,
113+
dash: dash,
114+
cache: cache,
115+
workdir: workdir,
116+
}
117+
118+
ctx, stop := context.WithCancel(context.Background())
119+
var wg sync.WaitGroup
120+
wg.Add(1)
121+
go func() {
122+
defer wg.Done()
123+
for {
124+
ok, err := s.poll(ctx)
125+
if err != nil {
126+
log.Error(err)
127+
dash.LogError("syz-agent", "%v", err)
128+
}
129+
var delay time.Duration
130+
if !ok {
131+
// Don't poll dashboard too often, if there are no jobs,
132+
// or errors are happenning.
133+
delay = 10 * time.Second
134+
}
135+
select {
136+
case <-ctx.Done():
137+
return
138+
case <-time.After(delay):
139+
}
140+
}
141+
}()
142+
143+
select {
144+
case <-shutdownPending:
145+
case <-updatePending:
146+
}
147+
stop()
148+
wg.Wait()
149+
150+
select {
151+
case <-shutdownPending:
152+
default:
153+
updater.UpdateAndRestart()
154+
}
155+
return nil
156+
}
157+
158+
type Server struct {
159+
cfg *Config
160+
dash *dashapi.Dashboard
161+
cache *aflow.Cache
162+
workdir string
163+
}
164+
165+
func (s *Server) poll(ctx context.Context) (
166+
bool, error) {
167+
req := &dashapi.AIJobPollReq{
168+
LLMModel: s.cfg.Model,
169+
CodeRevision: prog.GitRevision,
170+
}
171+
for _, flow := range aflow.Flows {
172+
req.Workflows = append(req.Workflows, dashapi.AIWorkflow{
173+
Type: flow.Type,
174+
Name: flow.Name,
175+
})
176+
}
177+
resp, err := s.dash.AIJobPoll(req)
178+
if err != nil {
179+
return false, err
180+
}
181+
if resp.ID == "" {
182+
return false, nil
183+
}
184+
doneReq := &dashapi.AIJobDoneReq{
185+
ID: resp.ID,
186+
}
187+
results, jobErr := s.executeJob(ctx, resp)
188+
doneReq.Results = results
189+
if jobErr != nil {
190+
doneReq.Error = jobErr.Error()
191+
}
192+
if err := s.dash.AIJobDone(doneReq); err != nil {
193+
return false, err
194+
}
195+
if jobErr != nil {
196+
return false, jobErr
197+
}
198+
return true, nil
199+
}
200+
201+
func (s *Server) executeJob(ctx context.Context, req *dashapi.AIJobPollResp) (map[string]any, error) {
202+
flow := aflow.Flows[req.Workflow]
203+
if flow == nil {
204+
return nil, fmt.Errorf("unsupported flow %q", req.Workflow)
205+
}
206+
inputs := map[string]any{
207+
"Syzkaller": osutil.Abs(filepath.FromSlash("syzkaller/current")),
208+
"CodesearchToolBin": s.cfg.CodesearchToolBin,
209+
"Image": s.cfg.Image,
210+
"Type": s.cfg.Type,
211+
"VM": s.cfg.VM,
212+
"FixedBaseCommit": s.cfg.FixedBaseCommit,
213+
}
214+
maps.Insert(inputs, maps.All(req.Args))
215+
onEvent := func(span *trajectory.Span) error {
216+
log.Logf(0, "%v", span)
217+
return s.dash.AITrajectoryLog(&dashapi.AITrajectoryReq{
218+
JobID: req.ID,
219+
Span: span,
220+
})
221+
}
222+
return flow.Execute(ctx, s.cfg.Model, s.workdir, inputs, s.cache, onEvent)
223+
}

0 commit comments

Comments
 (0)