mirror of
https://github.com/BaaRai/ARTEX.git
synced 2026-11-15 04:30:55 +08:00
Compare commits
8
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
cbb6f66c4b | ||
|
|
bba7989903 | ||
|
|
89ece8eb8d | ||
|
|
623e5e58a7 | ||
|
|
0777edfb8f | ||
|
|
67971ef7da | ||
|
|
57cac93d49 | ||
|
|
b269d3e590 |
@@ -0,0 +1,51 @@
|
||||
package db
|
||||
|
||||
import "context"
|
||||
|
||||
// SkillGap is one skill the agent asked for but the platform did not provide,
|
||||
// aggregated by skill, agent, and task category. A miss row is written when the
|
||||
// model names a skill that does not exist (skill_usage.found=false), so this
|
||||
// report is the evidence-backed capability backlog rather than a guess.
|
||||
type SkillGap struct {
|
||||
Skill string `json:"skill"`
|
||||
AgentKey string `json:"agent_key"`
|
||||
CategoryID int64 `json:"category_id"`
|
||||
Misses int `json:"misses"`
|
||||
Hits int `json:"hits"`
|
||||
LastSeen int64 `json:"last_seen"`
|
||||
}
|
||||
|
||||
// SkillGaps returns the aggregated miss report, highest miss count first.
|
||||
// Hits are the successful calls for the same key, so callers can compute a
|
||||
// miss rate without a second query.
|
||||
func (d *DB) SkillGaps(ctx context.Context, limit int) ([]SkillGap, error) {
|
||||
if limit <= 0 {
|
||||
limit = 50
|
||||
}
|
||||
rows, err := d.QueryContext(ctx, `
|
||||
SELECT su.skill,
|
||||
COALESCE(su.agent_key,''),
|
||||
COALESCE(t.category_id,0),
|
||||
COUNT(*) FILTER (WHERE NOT su.found) AS misses,
|
||||
COUNT(*) FILTER (WHERE su.found) AS hits,
|
||||
EXTRACT(EPOCH FROM MAX(su.ts))::bigint
|
||||
FROM skill_usage su
|
||||
LEFT JOIN tasks t ON t.id=su.task_id
|
||||
GROUP BY 1,2,3
|
||||
HAVING COUNT(*) FILTER (WHERE NOT su.found) > 0
|
||||
ORDER BY misses DESC, 6 DESC, 1
|
||||
LIMIT $1`, limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := []SkillGap{}
|
||||
for rows.Next() {
|
||||
var g SkillGap
|
||||
if err := rows.Scan(&g.Skill, &g.AgentKey, &g.CategoryID, &g.Misses, &g.Hits, &g.LastSeen); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, g)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
@@ -36,7 +36,6 @@ FLAG_RE = re.compile(r"(?:FLAG\{[0-9a-f]{64}\}|flag\{[0-9a-f-]{36}\})")
|
||||
CHALLENGE_RE = re.compile(r"DBEN-v1-\d{3}")
|
||||
|
||||
APP_SERVICE = "app"
|
||||
APP_CONTAINER_PORT = "8080"
|
||||
|
||||
|
||||
def random_flag() -> str:
|
||||
@@ -80,13 +79,33 @@ def compose_project(challenge: str, run_id: str) -> str:
|
||||
|
||||
|
||||
def host_port(output: str) -> int:
|
||||
"""Parse the host port out of `docker compose port app 8080` output."""
|
||||
"""Parse the host port out of `docker compose port` output."""
|
||||
line = output.strip().splitlines()[-1] if output.strip() else ""
|
||||
if ":" not in line:
|
||||
raise SystemExit(f"could not parse a host port from docker compose output: {output!r}")
|
||||
return int(line.rsplit(":", 1)[1])
|
||||
|
||||
|
||||
def container_host_port(project: str, cdir: pathlib.Path) -> int:
|
||||
"""Host port published by the app service, whatever container port the
|
||||
challenge declares (most challenges use 8080, some use 5000: never assume)."""
|
||||
res = _compose(project, cdir, ["ps", "-q", APP_SERVICE], timeout=60)
|
||||
cid = res.stdout.strip().splitlines()[0] if res.stdout.strip() else ""
|
||||
if not cid:
|
||||
raise SystemExit(f"no running {APP_SERVICE} container for project {project}")
|
||||
insp = subprocess.run(
|
||||
["docker", "inspect", "--format",
|
||||
"{{range $p, $conf := .NetworkSettings.Ports}}{{(index $conf 0).HostPort}}\n{{end}}",
|
||||
cid],
|
||||
capture_output=True, text=True, timeout=60)
|
||||
for token in insp.stdout.split():
|
||||
if token.isdigit():
|
||||
return int(token)
|
||||
raise SystemExit(
|
||||
f"no published host port on container {cid} "
|
||||
f"({insp.stderr.strip()[:200] or 'inspect returned no port mapping'})")
|
||||
|
||||
|
||||
def derive_ground_truth(repo: str, challenge: str) -> dict:
|
||||
"""Ground truth from the challenge's own benchmark.json.
|
||||
|
||||
@@ -168,8 +187,7 @@ def deploy(repo: str, challenge: str, flag: str, run_id: str) -> dict:
|
||||
if up.returncode != 0:
|
||||
print("[drost] up --wait failed; falling back to up -d and polling")
|
||||
_compose(project, cdir, ["up", "-d"], timeout=900)
|
||||
res = _compose(project, cdir, ["port", APP_SERVICE, APP_CONTAINER_PORT], timeout=60)
|
||||
url = f"http://127.0.0.1:{host_port(res.stdout)}"
|
||||
url = f"http://127.0.0.1:{container_host_port(project, cdir)}"
|
||||
for _ in range(60):
|
||||
try:
|
||||
urllib.request.urlopen(url, timeout=5)
|
||||
|
||||
@@ -189,18 +189,21 @@ def fetch_findings(api: Api, tid) -> list[dict]:
|
||||
out = []
|
||||
for f in items:
|
||||
fid = f.get("id") or f.get("node_id")
|
||||
# The detail route keys on finding_id (the findings-table row), not on
|
||||
# the exploration node id the list also carries; asking with the node
|
||||
# id 404s. Fall back to the node id so older payloads still resolve.
|
||||
did = f.get("finding_id") or fid
|
||||
detail = {}
|
||||
try:
|
||||
detail = api.get(f"/api/exploration/findings/{fid}")
|
||||
except SystemExit:
|
||||
raise
|
||||
except Exception:
|
||||
pass
|
||||
detail = api.get(f"/api/exploration/findings/{did}")
|
||||
except (SystemExit, Exception):
|
||||
pass # a missing detail must not kill the run; list fields remain
|
||||
report = detail.get("report") or f.get("report") or ""
|
||||
summary = f.get("summary") or detail.get("summary") or ""
|
||||
name = f.get("name") or f.get("title") or summary[:80]
|
||||
out.append({
|
||||
"id": fid,
|
||||
"finding_id": str(f.get("finding_id") or ""),
|
||||
"name": str(name)[:200],
|
||||
"severity": str(f.get("severity") or detail.get("severity") or "?"),
|
||||
"summary": str(summary)[:600],
|
||||
@@ -217,13 +220,14 @@ def deep_evidence_texts(api: Api, tid, findings: list[dict]) -> list[str]:
|
||||
common path."""
|
||||
texts = []
|
||||
for f in findings:
|
||||
did = f.get("finding_id") or f.get("id")
|
||||
try:
|
||||
texts.append(json.dumps(api.get(f"/api/exploration/findings/{f['id']}")))
|
||||
except Exception:
|
||||
texts.append(json.dumps(api.get(f"/api/exploration/findings/{did}")))
|
||||
except (SystemExit, Exception):
|
||||
continue
|
||||
try:
|
||||
texts.append(json.dumps(api.get(f"/api/exploration/nodes?task={tid}&kind=fact&size=200")))
|
||||
except Exception:
|
||||
except (SystemExit, Exception):
|
||||
pass
|
||||
try:
|
||||
r = urllib.request.Request(f"{api.base}/api/report?task={tid}")
|
||||
@@ -468,72 +472,83 @@ def main() -> None:
|
||||
gt = json.loads(gt_path.read_text())
|
||||
gt_findings = gt["findings"]
|
||||
|
||||
dep = deploy_corpus(args)
|
||||
corpus_url = dep["url"]
|
||||
flag = dep.get("flag")
|
||||
if args.deploy_only:
|
||||
print(f"[drost] deploy-only: up at {corpus_url}, flag {flag}")
|
||||
print(f"[drost] stop it later with: docker compose -p {dep['project']} down --remove-orphans"
|
||||
f" (cwd {dep['challenge_dir']})")
|
||||
return
|
||||
|
||||
task_url = corpus_url
|
||||
if args.target_host != "127.0.0.1":
|
||||
task_url = f"http://{args.target_host}:{corpus_url.rsplit(':', 1)[1]}"
|
||||
print(f"[corpus] task target rewritten for ARTEX reachability: {task_url}")
|
||||
|
||||
api = Api(args.artex_url, token)
|
||||
task_info = create_task(api, args, task_url, gt)
|
||||
status = wait_task(api, task_info["id"], args.task_timeout + 1800)
|
||||
findings = fetch_findings(api, task_info["id"])
|
||||
|
||||
invalid_ids = [f["id"] for f in findings if not (f["summary"].strip() or f["report"].strip())]
|
||||
usable = [f for f in findings if f["id"] not in invalid_ids]
|
||||
|
||||
flag_captured = None
|
||||
if flag:
|
||||
# Objective oracle first pass over the already-fetched fields; the deep
|
||||
# scan (full details, facts, report) runs only before declaring a miss.
|
||||
flag_captured = drost.flag_in_texts(flag, drost.finding_texts(findings))
|
||||
if not flag_captured:
|
||||
flag_captured = drost.flag_in_texts(flag, deep_evidence_texts(api, task_info["id"], findings))
|
||||
print(f"[flag] captured: {flag_captured}")
|
||||
|
||||
if judge_enabled:
|
||||
if not usable:
|
||||
print("[judge] no usable findings; scoring zeros")
|
||||
verdicts = [{"gt_id": g["id"], "detected": False, "matched_finding_id": None,
|
||||
"gates": {"component": False, "class": False, "keywords": False, "proof": False},
|
||||
"reason": "no usable scanner findings"} for g in gt_findings]
|
||||
else:
|
||||
print("[judge] running falsification battery")
|
||||
judge_battery(args)
|
||||
print("[judge] grading")
|
||||
verdicts = run_judge(args, gt_findings, usable)
|
||||
result = tally(gt_findings, verdicts, findings, invalid_ids)
|
||||
else:
|
||||
print("[judge] flag-only run: no judge configured, analysis gates not measured")
|
||||
verdicts = None
|
||||
result = {
|
||||
"mode": "flag-only",
|
||||
"scanner_findings_total": len(findings),
|
||||
"invalid_findings_excluded": invalid_ids,
|
||||
"discovery_candidates": [{"id": f["id"], "name": f["name"]} for f in usable],
|
||||
"note": "analysis gates not measured: no judge configured",
|
||||
}
|
||||
if flag:
|
||||
result["flag_captured"] = flag_captured
|
||||
result["flag_used"] = flag
|
||||
|
||||
out_dir = pathlib.Path("benchmark-results") / args.corpus / args.run_id
|
||||
write_report(out_dir, args.corpus, args, task_info, status, findings, verdicts, result)
|
||||
print(json.dumps(result, indent=2)[:1500])
|
||||
|
||||
# drost project/dir are deterministic, so they are known before deploy and
|
||||
# the finally below can clean up even when the deploy itself fails halfway.
|
||||
drost_project = drost_cdir = None
|
||||
if args.corpus_type == "drost":
|
||||
if not args.keep:
|
||||
drost.teardown(dep["project"], dep["challenge_dir"])
|
||||
elif not (args.keep or args.corpus_url):
|
||||
subprocess.run(["docker", "rm", "-f", f"bench-{args.corpus}-{args.run_id}"], capture_output=True)
|
||||
drost_project = drost.compose_project(args.drost_id, args.run_id)
|
||||
drost_cdir = str(drost.challenge_dir(args.drost_repo, args.drost_id))
|
||||
|
||||
try:
|
||||
dep = deploy_corpus(args)
|
||||
corpus_url = dep["url"]
|
||||
flag = dep.get("flag")
|
||||
if args.deploy_only:
|
||||
print(f"[drost] deploy-only: up at {corpus_url}, flag {flag}")
|
||||
print(f"[drost] stop it later with: docker compose -p {dep['project']} down --remove-orphans"
|
||||
f" (cwd {dep['challenge_dir']})")
|
||||
return
|
||||
|
||||
task_url = corpus_url
|
||||
if args.target_host != "127.0.0.1":
|
||||
task_url = f"http://{args.target_host}:{corpus_url.rsplit(':', 1)[1]}"
|
||||
print(f"[corpus] task target rewritten for ARTEX reachability: {task_url}")
|
||||
|
||||
api = Api(args.artex_url, token)
|
||||
task_info = create_task(api, args, task_url, gt)
|
||||
status = wait_task(api, task_info["id"], args.task_timeout + 1800)
|
||||
findings = fetch_findings(api, task_info["id"])
|
||||
|
||||
invalid_ids = [f["id"] for f in findings if not (f["summary"].strip() or f["report"].strip())]
|
||||
usable = [f for f in findings if f["id"] not in invalid_ids]
|
||||
|
||||
flag_captured = None
|
||||
if flag:
|
||||
# Objective oracle first pass over the already-fetched fields; the deep
|
||||
# scan (full details, facts, report) runs only before declaring a miss.
|
||||
flag_captured = drost.flag_in_texts(flag, drost.finding_texts(findings))
|
||||
if not flag_captured:
|
||||
flag_captured = drost.flag_in_texts(flag, deep_evidence_texts(api, task_info["id"], findings))
|
||||
print(f"[flag] captured: {flag_captured}")
|
||||
|
||||
if judge_enabled:
|
||||
if not usable:
|
||||
print("[judge] no usable findings; scoring zeros")
|
||||
verdicts = [{"gt_id": g["id"], "detected": False, "matched_finding_id": None,
|
||||
"gates": {"component": False, "class": False, "keywords": False, "proof": False},
|
||||
"reason": "no usable scanner findings"} for g in gt_findings]
|
||||
else:
|
||||
print("[judge] running falsification battery")
|
||||
judge_battery(args)
|
||||
print("[judge] grading")
|
||||
verdicts = run_judge(args, gt_findings, usable)
|
||||
result = tally(gt_findings, verdicts, findings, invalid_ids)
|
||||
else:
|
||||
print("[judge] flag-only run: no judge configured, analysis gates not measured")
|
||||
verdicts = None
|
||||
result = {
|
||||
"mode": "flag-only",
|
||||
"scanner_findings_total": len(findings),
|
||||
"invalid_findings_excluded": invalid_ids,
|
||||
"discovery_candidates": [{"id": f["id"], "name": f["name"]} for f in usable],
|
||||
"note": "analysis gates not measured: no judge configured",
|
||||
}
|
||||
if flag:
|
||||
result["flag_captured"] = flag_captured
|
||||
result["flag_used"] = flag
|
||||
|
||||
out_dir = pathlib.Path("benchmark-results") / args.corpus / args.run_id
|
||||
write_report(out_dir, args.corpus, args, task_info, status, findings, verdicts, result)
|
||||
print(json.dumps(result, indent=2)[:1500])
|
||||
finally:
|
||||
# The corpus is torn down even when a step fails halfway: a crash in
|
||||
# deploy, scoring or reporting must not leave the challenge running
|
||||
# (--keep and --deploy-only opt out deliberately).
|
||||
if args.corpus_type == "drost":
|
||||
if not (args.keep or args.deploy_only):
|
||||
drost.teardown(drost_project, drost_cdir)
|
||||
elif not (args.keep or args.corpus_url):
|
||||
subprocess.run(["docker", "rm", "-f", f"bench-{args.corpus}-{args.run_id}"], capture_output=True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
+34
-18
@@ -673,6 +673,7 @@ func (s *Server) Handler() http.Handler {
|
||||
|
||||
mux.HandleFunc("GET /api/tasks", s.listTasks)
|
||||
mux.HandleFunc("POST /api/tasks", s.createTask)
|
||||
mux.HandleFunc("GET /api/skill-gaps", s.skillGaps)
|
||||
mux.HandleFunc("GET /api/task-categories", s.pgListTaskCategories)
|
||||
mux.HandleFunc("POST /api/task-categories", s.pgCreateTaskCategory)
|
||||
mux.HandleFunc("PATCH /api/task-categories/{id}", s.pgRenameTaskCategory)
|
||||
@@ -1044,34 +1045,44 @@ func (s *Server) listTasks(w http.ResponseWriter, r *http.Request) {
|
||||
active = t.ID
|
||||
}
|
||||
list := s.m.List()
|
||||
metrics, _ := s.m.PG().TaskListMetricsAll()
|
||||
metrics, err := s.m.PG().TaskListMetricsAll()
|
||||
if err != nil {
|
||||
log.Printf("[tasks] list metrics query failed — metrics zeroed: %v", err)
|
||||
}
|
||||
archiveBlockers, _ := s.m.PG().TaskArchiveBlockers()
|
||||
dtos := make([]TaskDTO, 0, len(list))
|
||||
for _, t := range list {
|
||||
dto := taskDTO(t, s.resolvedTaskStatus(t))
|
||||
applyTaskArchiveBlocker(&dto, archiveBlockers)
|
||||
metric := metrics[t.ExpID]
|
||||
dto.Tokens = tokenTotalDTO(metric.Tokens)
|
||||
// prefer the live in-memory heartbeat (fresher) and fall back to the
|
||||
// persisted max activity time (survives restarts) for run-duration display.
|
||||
dto.LastActivity = metric.LastActivity
|
||||
if live := s.engine.LastActivity(t.ID); live > dto.LastActivity {
|
||||
dto.LastActivity = live
|
||||
}
|
||||
dto.GoalsTotal = metric.Goals.Total
|
||||
dto.GoalsMet = metric.Goals.Met
|
||||
dto.InFlight = metric.RunningIntents
|
||||
dto.Findings = FindingSeverityDTO{
|
||||
Critical: metric.Findings.Critical,
|
||||
High: metric.Findings.High,
|
||||
Medium: metric.Findings.Medium,
|
||||
Low: metric.Findings.Low,
|
||||
}
|
||||
s.applyTaskMetrics(&dto, t, metrics[t.ExpID])
|
||||
dtos = append(dtos, dto)
|
||||
}
|
||||
writeJSON(w, 200, map[string]any{"tasks": dtos, "active": active})
|
||||
}
|
||||
|
||||
// applyTaskMetrics fills the aggregate DTO fields (tokens, goal progress,
|
||||
// running intents, findings, last activity) from the list-metrics query so the
|
||||
// detail and list endpoints report identical numbers. A zero metric leaves the
|
||||
// DTO zeroed, which is why callers log the query error instead of assuming it.
|
||||
func (s *Server) applyTaskMetrics(dto *TaskDTO, t *Task, metric db.TaskListMetrics) {
|
||||
dto.Tokens = tokenTotalDTO(metric.Tokens)
|
||||
// prefer the live in-memory heartbeat (fresher) and fall back to the
|
||||
// persisted max activity time (survives restarts) for run-duration display.
|
||||
dto.LastActivity = metric.LastActivity
|
||||
if live := s.engine.LastActivity(t.ID); live > dto.LastActivity {
|
||||
dto.LastActivity = live
|
||||
}
|
||||
dto.GoalsTotal = metric.Goals.Total
|
||||
dto.GoalsMet = metric.Goals.Met
|
||||
dto.InFlight = metric.RunningIntents
|
||||
dto.Findings = FindingSeverityDTO{
|
||||
Critical: metric.Findings.Critical,
|
||||
High: metric.Findings.High,
|
||||
Medium: metric.Findings.Medium,
|
||||
Low: metric.Findings.Low,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) resolvedTaskStatus(t *Task) string {
|
||||
if t == nil {
|
||||
return "created"
|
||||
@@ -1798,6 +1809,11 @@ func (s *Server) getTask(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
dto := taskDTO(t, s.resolvedTaskStatus(t))
|
||||
metrics, err := s.m.PG().TaskListMetricsAll()
|
||||
if err != nil {
|
||||
log.Printf("[tasks] detail metrics query failed for %s — metrics zeroed: %v", t.ID, err)
|
||||
}
|
||||
s.applyTaskMetrics(&dto, t, metrics[t.ExpID])
|
||||
archiveBlockers, _ := s.m.PG().TaskArchiveBlockers()
|
||||
applyTaskArchiveBlocker(&dto, archiveBlockers)
|
||||
writeJSON(w, 200, dto)
|
||||
|
||||
@@ -0,0 +1,30 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"log"
|
||||
"net/http"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
// skillGaps reports skills the agent asked for but the platform did not
|
||||
// provide, aggregated by skill, agent, and task category. The source is the
|
||||
// existing skill_usage miss rows, so the backlog is evidence-backed rather than
|
||||
// inferred. Read-only: it creates no skill and changes no prompt.
|
||||
func (s *Server) skillGaps(w http.ResponseWriter, r *http.Request) {
|
||||
limit := 50
|
||||
if raw := r.URL.Query().Get("limit"); raw != "" {
|
||||
n, err := strconv.Atoi(raw)
|
||||
if err != nil || n < 1 || n > 500 {
|
||||
writeErr(w, 400, "limit must be an integer between 1 and 500")
|
||||
return
|
||||
}
|
||||
limit = n
|
||||
}
|
||||
gaps, err := s.m.PG().SkillGaps(r.Context(), limit)
|
||||
if err != nil {
|
||||
log.Printf("[skills] gap query failed: %v", err)
|
||||
writeErr(w, 500, "skill gap query failed")
|
||||
return
|
||||
}
|
||||
writeJSON(w, 200, map[string]any{"gaps": gaps, "count": len(gaps)})
|
||||
}
|
||||
@@ -0,0 +1,107 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strconv"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestSkillGapsReport(t *testing.T) {
|
||||
m, err := NewManager(t.TempDir(), "")
|
||||
if err != nil {
|
||||
t.Skipf("postgres unavailable (%v) — skipping", err)
|
||||
}
|
||||
defer m.Close()
|
||||
task, err := m.CreateTask("skill gaps", "goal", nil, 0, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
taskID, _ := strconv.ParseInt(task.ID, 10, 64)
|
||||
defer func() { _ = m.pg.DeleteTask(taskID) }()
|
||||
defer func() { _, _ = m.pg.Exec(`DELETE FROM skill_usage WHERE task_id=$1`, taskID) }()
|
||||
|
||||
if _, err := m.pg.Exec(`
|
||||
INSERT INTO skill_usage(skill,agent_key,task_id,found) VALUES
|
||||
('rsi-test-graphql','worker',$1,false),
|
||||
('rsi-test-graphql','worker',$1,false),
|
||||
('rsi-test-graphql','worker',$1,true),
|
||||
('rsi-test-cloud','worker',$1,false),
|
||||
('rsi-test-known','worker',$1,true)`, taskID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
s := New(context.Background(), m, t.TempDir(), t.TempDir(), t.TempDir())
|
||||
token, err := signJWT(s.jwtKey)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
get := func(path string) *httptest.ResponseRecorder {
|
||||
req := httptest.NewRequest(http.MethodGet, path, nil)
|
||||
req.Header.Set("Authorization", "Bearer "+token)
|
||||
rec := httptest.NewRecorder()
|
||||
s.Handler().ServeHTTP(rec, req)
|
||||
return rec
|
||||
}
|
||||
|
||||
rec := get("/api/skill-gaps")
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status=%d body=%s", rec.Code, rec.Body.String())
|
||||
}
|
||||
var out struct {
|
||||
Gaps []struct {
|
||||
Skill string `json:"skill"`
|
||||
AgentKey string `json:"agent_key"`
|
||||
Misses int `json:"misses"`
|
||||
Hits int `json:"hits"`
|
||||
} `json:"gaps"`
|
||||
Count int `json:"count"`
|
||||
}
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
bySkill := map[string]struct {
|
||||
misses, hits int
|
||||
agent string
|
||||
}{}
|
||||
for _, g := range out.Gaps {
|
||||
bySkill[g.Skill] = struct {
|
||||
misses, hits int
|
||||
agent string
|
||||
}{g.Misses, g.Hits, g.AgentKey}
|
||||
}
|
||||
if got := bySkill["rsi-test-graphql"]; got.misses != 2 || got.hits != 1 || got.agent != "worker" {
|
||||
t.Fatalf("graphql gap=%+v want misses=2 hits=1 agent=worker", got)
|
||||
}
|
||||
if got := bySkill["rsi-test-cloud"]; got.misses != 1 {
|
||||
t.Fatalf("cloud gap=%+v want misses=1", got)
|
||||
}
|
||||
if _, ok := bySkill["rsi-test-known"]; ok {
|
||||
t.Fatal("a hit-only skill must not appear in the miss report")
|
||||
}
|
||||
first := out.Gaps[0].Skill
|
||||
if first != "rsi-test-graphql" {
|
||||
t.Fatalf("ordering: first=%q want rsi-test-graphql (highest misses)", first)
|
||||
}
|
||||
|
||||
rec = get("/api/skill-gaps?limit=1")
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("limit status=%d body=%s", rec.Code, rec.Body.String())
|
||||
}
|
||||
var limited struct {
|
||||
Gaps []struct {
|
||||
Skill string `json:"skill"`
|
||||
} `json:"gaps"`
|
||||
}
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &limited); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(limited.Gaps) != 1 || limited.Gaps[0].Skill != "rsi-test-graphql" {
|
||||
t.Fatalf("limit=1 returned %+v", limited.Gaps)
|
||||
}
|
||||
if rec := get("/api/skill-gaps?limit=abc"); rec.Code != http.StatusBadRequest {
|
||||
t.Fatalf("bad limit status=%d want 400", rec.Code)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,113 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strconv"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestTaskDetailReturnsListMetrics(t *testing.T) {
|
||||
m, err := NewManager(t.TempDir(), "")
|
||||
if err != nil {
|
||||
t.Skipf("postgres unavailable (%v) — skipping", err)
|
||||
}
|
||||
defer m.Close()
|
||||
task, err := m.CreateTask("detail metrics", "goal", nil, 0, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
taskID, _ := strconv.ParseInt(task.ID, 10, 64)
|
||||
defer func() { _ = m.pg.DeleteTask(taskID) }()
|
||||
|
||||
// Seed the aggregates the metrics query reads: one result-activity with
|
||||
// tokens, one non-result row that must be excluded, and two goals (1 met).
|
||||
if _, err := m.pg.Exec(`
|
||||
INSERT INTO activity(
|
||||
exploration_id, kind, input_tokens, output_tokens,
|
||||
cache_read_tokens, cache_write_tokens, created_at
|
||||
) VALUES ($1,'result',11,7,3,2,now()), ($1,'tool_result',999,999,999,999,now())`,
|
||||
task.ExpID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := m.pg.Exec(`
|
||||
INSERT INTO exploration_nodes(exploration_id, kind, payload, state)
|
||||
VALUES ($1,'goal','{}','met'), ($1,'goal','{}','open')`, task.ExpID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
s := New(context.Background(), m, t.TempDir(), t.TempDir(), t.TempDir())
|
||||
token, err := signJWT(s.jwtKey)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
var probe struct {
|
||||
Tokens struct {
|
||||
InputTokens int `json:"input_tokens"`
|
||||
OutputTokens int `json:"output_tokens"`
|
||||
CacheReadTokens int `json:"cache_read_tokens"`
|
||||
CacheWriteTokens int `json:"cache_write_tokens"`
|
||||
} `json:"tokens"`
|
||||
GoalsMet int `json:"goals_met"`
|
||||
GoalsTotal int `json:"goals_total"`
|
||||
}
|
||||
req := httptest.NewRequest(http.MethodGet, "/api/tasks/"+task.ID, nil)
|
||||
req.Header.Set("Authorization", "Bearer "+token)
|
||||
rec := httptest.NewRecorder()
|
||||
s.Handler().ServeHTTP(rec, req)
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("detail status=%d body=%s", rec.Code, rec.Body.String())
|
||||
}
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &probe); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if probe.Tokens.InputTokens != 11 || probe.Tokens.OutputTokens != 7 ||
|
||||
probe.Tokens.CacheReadTokens != 3 || probe.Tokens.CacheWriteTokens != 2 {
|
||||
t.Fatalf("detail tokens not filled from metrics: %+v", probe.Tokens)
|
||||
}
|
||||
if probe.GoalsMet != 1 || probe.GoalsTotal != 2 {
|
||||
t.Fatalf("detail goal progress not filled from metrics: %d/%d", probe.GoalsMet, probe.GoalsTotal)
|
||||
}
|
||||
|
||||
// The list endpoint must report the same numbers for the same task.
|
||||
listReq := httptest.NewRequest(http.MethodGet, "/api/tasks", nil)
|
||||
listReq.Header.Set("Authorization", "Bearer "+token)
|
||||
listRec := httptest.NewRecorder()
|
||||
s.Handler().ServeHTTP(listRec, listReq)
|
||||
if listRec.Code != http.StatusOK {
|
||||
t.Fatalf("list status=%d body=%s", listRec.Code, listRec.Body.String())
|
||||
}
|
||||
var listed struct {
|
||||
Tasks []struct {
|
||||
ID string `json:"id"`
|
||||
Tokens struct {
|
||||
InputTokens int `json:"input_tokens"`
|
||||
OutputTokens int `json:"output_tokens"`
|
||||
} `json:"tokens"`
|
||||
GoalsMet int `json:"goals_met"`
|
||||
GoalsTotal int `json:"goals_total"`
|
||||
} `json:"tasks"`
|
||||
}
|
||||
if err := json.Unmarshal(listRec.Body.Bytes(), &listed); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
found := false
|
||||
for _, lt := range listed.Tasks {
|
||||
if lt.ID != task.ID {
|
||||
continue
|
||||
}
|
||||
found = true
|
||||
if lt.Tokens.InputTokens != probe.Tokens.InputTokens ||
|
||||
lt.Tokens.OutputTokens != probe.Tokens.OutputTokens ||
|
||||
lt.GoalsMet != probe.GoalsMet || lt.GoalsTotal != probe.GoalsTotal {
|
||||
t.Fatalf("list and detail metrics disagree: list=%+v/%d detail=%+v/%d",
|
||||
lt.Tokens, lt.GoalsMet, probe.Tokens, probe.GoalsTotal)
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Fatalf("task %s missing from list response", task.ID)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user