Skip to content

Commit 5335465

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 1e5e646 commit 5335465

File tree

3 files changed

+230
-2
lines changed

3 files changed

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

0 commit comments

Comments
 (0)