mirror of
https://github.com/esengine/DeepSeek-Reasonix.git
synced 2026-11-07 10:45:59 -10:00
fix(cli): restore 1.x's stream-json and events-jsonl line contract
1.x's `run --output-format stream-json` stamped every line with sessionId, turnId, seq and status, opened the turn with turn_status (queued) and user_message, reported tool_started between a call's dispatch and its result, and closed with turn_done; --events-jsonl carried the same lifecycle records without content. Studio's synchronous controller path publishes no turn lifecycle, so both streams lacked all four kinds and the envelope, and they leaked host-internal kinds (workspace_lease and friends) that no 1.x consumer knows. The run's output sink now supplies the lifecycle for the turn it drives and stamps the envelope; only the kinds 1.x emitted reach either stream. tool_started comes from a new host-only Tool.Executed on ToolResult, set when the call cleared every gate and ran, so a refused call reports none. The stats recorder swallowed RecordTurnCompletion instead of passing it to the sink it wraps, so the run sink never learned a turn had finished: a failed run reported num_turns 0. It now forwards the signal once (a TurnDone event still records without a second completion). Contract tests compare a real run against v1.39.5 output kept under testdata/v1-run: lifecycle order, envelope keys, kind vocabulary.
This commit is contained in:
1 parent
8016f8be98
commit
fb6a82c8e3
14 files changed
+566
-10
No files matched your search
+13
@@ -210,6 +210,16 @@ reasonix -p "summarize the diff" --output-format json
|
||||
reasonix run "run the tests" --output-format stream-json
|
||||
```
|
||||
|
||||
`stream-json` lines before the result follow the 1.x contract:
|
||||
|
||||
- Each carries `sessionId`, `turnId`, `seq` (from 1) and `status`.
|
||||
- The turn opens with `turn_status` (`status: "queued"`) and `user_message`.
|
||||
- A call that cleared every gate and ran reports `tool_started` before its
|
||||
`tool_result`.
|
||||
- The turn closes with `turn_done`: `completed`, `failed` or `interrupted`.
|
||||
- Only kinds 1.x emitted appear; host-internal state such as workspace leases
|
||||
stays off the stream.
|
||||
|
||||
The final structured object has this shape:
|
||||
|
||||
```json
|
||||
@@ -288,6 +298,9 @@ Every line has `schema_version`, `sequence`, and `kind`; the final line is
|
||||
`--output-format stream-json` contract and cannot be combined with
|
||||
`--output-format`.
|
||||
|
||||
Its lifecycle records match `stream-json`'s (`turn_status`, `user_message`,
|
||||
`tool_started`, `turn_done`) without their content.
|
||||
|
||||
The following read-only commands expose persisted state without transcript,
|
||||
label, command, output, path, PID, or host-name content. Here, read-only means
|
||||
the commands do not mutate transcript, runtime, recovery, or query state. The
|
||||
|
||||
+10
-1
@@ -182,6 +182,14 @@ reasonix -p "总结 diff" --output-format json
|
||||
reasonix run "运行测试" --output-format stream-json
|
||||
```
|
||||
|
||||
`stream-json` 在结果对象之前的各行遵循 1.x 的约定:
|
||||
|
||||
- 每行都带 `sessionId`、`turnId`、`seq`(从 1 开始)和 `status`。
|
||||
- 回合以 `turn_status`(`status: "queued"`)和 `user_message` 开始。
|
||||
- 通过全部关卡并真正执行的工具调用,在 `tool_result` 之前先报 `tool_started`。
|
||||
- 回合以 `turn_done` 结束:`completed`、`failed` 或 `interrupted`。
|
||||
- 只输出 1.x 输出过的事件种类,工作区租约之类的宿主内部状态不进入这条流。
|
||||
|
||||
最终结构化对象的格式如下:
|
||||
|
||||
```json
|
||||
@@ -250,7 +258,8 @@ reasonix run --events-jsonl "运行 focused tests"
|
||||
```
|
||||
|
||||
每行都包含 `schema_version`、`sequence` 和 `kind`,最后一行为
|
||||
`kind: "run_done"`。`--events-jsonl` 与包含更多内容的
|
||||
`kind: "run_done"`。生命周期记录与 `stream-json` 相同(`turn_status`、`user_message`、
|
||||
`tool_started`、`turn_done`),但不带内容。`--events-jsonl` 与包含更多内容的
|
||||
`--output-format stream-json` 是两个独立契约,不能和 `--output-format`
|
||||
组合使用。
|
||||
|
||||
|
||||
@@ -297,6 +297,8 @@ type Tool struct {
|
||||
WorkspaceMutation bool
|
||||
WorkspacePaths []string
|
||||
WorkspaceAllPaths bool
|
||||
// Executed marks a ToolResult whose call cleared every gate and ran. Host-only.
|
||||
Executed bool
|
||||
}
|
||||
|
||||
// ShellExecution mirrors tool.ShellExecution for event sinks without importing
|
||||
|
||||
@@ -545,7 +545,7 @@ func runAgent(args []string, version string) int {
|
||||
return 1
|
||||
}
|
||||
|
||||
recordTrajectoryHeader(chain.trajectory, ctrl, version)
|
||||
chain.begin(ctrl, version, prompt)
|
||||
runErr := ctrl.Run(ctx, prompt)
|
||||
reporter.RecordRecovery(ctrl.DrainRecoveryMetrics())
|
||||
completion := classifyRunCompletion(runErr)
|
||||
|
||||
@@ -174,6 +174,7 @@ type runOutputSink struct {
|
||||
nextMachineToolID uint64
|
||||
nextMachineToolName uint64
|
||||
permissions runPermissionRecord
|
||||
turn runTurnEnvelope
|
||||
err error
|
||||
}
|
||||
|
||||
@@ -256,11 +257,8 @@ func (s *runOutputSink) Emit(e event.Event) {
|
||||
if s.format == runOutputText && e.Kind == event.Notice && (e.Level == event.LevelWarn || closesWarning(e.Code)) {
|
||||
s.writeDiagnostic(e)
|
||||
}
|
||||
if s.format == runOutputStreamJSON && s.err == nil {
|
||||
s.err = s.encoder.Encode(eventwire.ToWire(e))
|
||||
} else if s.format == runOutputEventsJSONL && s.err == nil {
|
||||
s.sequence++
|
||||
s.err = s.encoder.Encode(s.machineEventRecordFor(e, s.sequence))
|
||||
if s.format == runOutputStreamJSON || s.format == runOutputEventsJSONL {
|
||||
s.writeStreamEvent(e)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -302,6 +300,10 @@ func (s *runOutputSink) Finalize(sessionID string, started time.Time, runErr err
|
||||
return s.err
|
||||
}
|
||||
completion := classifyRunCompletion(runErr)
|
||||
s.finishTurn(runErr, completion)
|
||||
if s.err != nil {
|
||||
return s.err
|
||||
}
|
||||
if s.format == runOutputEventsJSONL {
|
||||
s.sequence++
|
||||
turns := s.turns
|
||||
|
||||
@@ -12,6 +12,7 @@ import (
|
||||
"reasonix/internal/platform/telemetry"
|
||||
"reasonix/internal/runtime/agent"
|
||||
"reasonix/internal/session/control"
|
||||
"reasonix/internal/state/sessionstore"
|
||||
"reasonix/internal/state/trajectory"
|
||||
)
|
||||
|
||||
@@ -68,6 +69,14 @@ func buildRunSink(format runOutputFormat, printOnly, showThinking bool, metricsP
|
||||
return chain, nil
|
||||
}
|
||||
|
||||
// begin marks the controller bound and the prompt about to be submitted.
|
||||
func (c runSinkChain) begin(ctrl *control.Controller, version, prompt string) {
|
||||
recordTrajectoryHeader(c.trajectory, ctrl, version)
|
||||
if c.resultOutput != nil {
|
||||
c.resultOutput.BeginTurn(sessionstore.BranchID(ctrl.SessionPath()), prompt)
|
||||
}
|
||||
}
|
||||
|
||||
// recordTrajectoryHeader persists the request-side prefix the event stream
|
||||
// cannot carry. The hashes come from the same capture the cache diagnostics
|
||||
// report, so a reader can prove the header belongs to the rounds it precedes
|
||||
|
||||
@@ -0,0 +1,289 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"reasonix/internal/base/testenv"
|
||||
"reasonix/internal/contract/eventwire"
|
||||
)
|
||||
|
||||
// testdata/v1-run holds `reasonix run` v1.39.5 output for one tool call and an
|
||||
// answer. A run here keeps its lifecycle order, per-line envelope, and kind set.
|
||||
|
||||
var runLifecycleKinds = []string{"turn_status", "turn_started", "user_message", "tool_started", "tool_result", "turn_done"}
|
||||
|
||||
type streamRecord map[string]any
|
||||
|
||||
func (r streamRecord) kind() string { s, _ := r["kind"].(string); return s }
|
||||
|
||||
func readStreamRecords(t *testing.T, text string) []streamRecord {
|
||||
t.Helper()
|
||||
var out []streamRecord
|
||||
sc := bufio.NewScanner(strings.NewReader(text))
|
||||
sc.Buffer(make([]byte, 1<<20), 1<<24)
|
||||
for sc.Scan() {
|
||||
line := strings.TrimSpace(sc.Text())
|
||||
if line == "" {
|
||||
continue
|
||||
}
|
||||
var rec streamRecord
|
||||
if err := json.Unmarshal([]byte(line), &rec); err != nil {
|
||||
t.Fatalf("line is not JSON: %v\n%s", err, line)
|
||||
}
|
||||
out = append(out, rec)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func readV1Sample(t *testing.T, name string) []streamRecord {
|
||||
t.Helper()
|
||||
data, err := os.ReadFile(filepath.Join("testdata", "v1-run", name))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return readStreamRecords(t, string(data))
|
||||
}
|
||||
|
||||
func lifecycleSkeleton(records []streamRecord) []string {
|
||||
var out []string
|
||||
for _, r := range records {
|
||||
if slices.Contains(runLifecycleKinds, r.kind()) {
|
||||
out = append(out, r.kind())
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func streamVocabulary() map[string]bool {
|
||||
names := map[string]bool{"result": true, "run_done": true}
|
||||
for _, k := range runLifecycleKinds {
|
||||
names[k] = true
|
||||
}
|
||||
for kind := range runStreamKinds {
|
||||
name, _ := eventwire.KindName(kind)
|
||||
names[name] = true
|
||||
}
|
||||
return names
|
||||
}
|
||||
|
||||
// runToolCallFixture answers the first request with one call to name(args) and
|
||||
// every later one with "done".
|
||||
func runToolCallFixture(t *testing.T, name, args string) string {
|
||||
t.Helper()
|
||||
var mu sync.Mutex
|
||||
turn := 0
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
_, _ = io.Copy(io.Discard, r.Body)
|
||||
mu.Lock()
|
||||
turn++
|
||||
n := turn
|
||||
mu.Unlock()
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
if n == 1 {
|
||||
call, _ := json.Marshal(map[string]any{"choices": []any{map[string]any{"index": 0, "delta": map[string]any{
|
||||
"tool_calls": []any{map[string]any{"index": 0, "id": "call_1", "type": "function",
|
||||
"function": map[string]any{"name": name, "arguments": args}}},
|
||||
}}}})
|
||||
_, _ = fmt.Fprintf(w, "data: %s\n\n", call)
|
||||
_, _ = io.WriteString(w, `data: {"choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":10,"completion_tokens":2,"total_tokens":12}}`+"\n\ndata: [DONE]\n\n")
|
||||
return
|
||||
}
|
||||
_, _ = io.WriteString(w, `data: {"choices":[{"index":0,"delta":{"content":"done"},"finish_reason":"stop"}],"usage":{"prompt_tokens":10,"completion_tokens":2,"total_tokens":12}}`+"\n\ndata: [DONE]\n\n")
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
isolateCLIConfigHome(t)
|
||||
home := testenv.TempDir(t)
|
||||
t.Setenv("REASONIX_HOME", home)
|
||||
t.Setenv("RUN_FLAGS_FAKE_KEY", "k")
|
||||
writeTestFile(t, filepath.Join(home, "config.toml"), fmt.Sprintf(`default_model = "fake"
|
||||
|
||||
[[providers]]
|
||||
name = "fake"
|
||||
kind = "openai"
|
||||
base_url = %q
|
||||
model = "fake-model"
|
||||
api_key_env = "RUN_FLAGS_FAKE_KEY"
|
||||
`, srv.URL), 0o644)
|
||||
dir := testenv.TempDir(t)
|
||||
t.Chdir(dir)
|
||||
writeTestFile(t, filepath.Join(dir, "note.txt"), "hello\n", 0o644)
|
||||
return dir
|
||||
}
|
||||
|
||||
func runCLIStream(t *testing.T, argv ...string) []streamRecord {
|
||||
t.Helper()
|
||||
var code int
|
||||
stdout, stderr := captureDrainedOutput(t, func() { code = Run(argv, "test") })
|
||||
if code != 0 {
|
||||
t.Fatalf("Run(%q) exited %d\nstdout:\n%s\nstderr:\n%s", argv, code, stdout, stderr)
|
||||
}
|
||||
return readStreamRecords(t, stdout)
|
||||
}
|
||||
|
||||
func TestRunStreamJSONKeeps1xLineContract(t *testing.T) {
|
||||
v1 := readV1Sample(t, "stream-json-write-file.jsonl")
|
||||
runToolCallFixture(t, "read_file", `{"path":"note.txt"}`)
|
||||
got := runCLIStream(t, "run", "--output-format", "stream-json", "READ note.txt")
|
||||
|
||||
if want, have := lifecycleSkeleton(v1), lifecycleSkeleton(got); !slices.Equal(want, have) {
|
||||
t.Fatalf("lifecycle records = %v, want 1.x's %v", have, want)
|
||||
}
|
||||
if last := got[len(got)-1]; last.kind() != "" || last["type"] != "result" || last["is_error"] != false || last["num_turns"] != float64(1) {
|
||||
t.Fatalf("last line = %v, want a successful result with num_turns 1", last)
|
||||
}
|
||||
events := got[:len(got)-1]
|
||||
vocab := streamVocabulary()
|
||||
var session, turn string
|
||||
for i, r := range events {
|
||||
for _, key := range []string{"sessionId", "turnId", "seq", "status"} {
|
||||
if _, ok := v1[0][key]; ok && r[key] == nil {
|
||||
t.Fatalf("line %d (%s) lacks 1.x envelope key %q: %v", i, r.kind(), key, r)
|
||||
}
|
||||
}
|
||||
if r["seq"] != float64(i+1) {
|
||||
t.Fatalf("line %d seq = %v, want %d", i, r["seq"], i+1)
|
||||
}
|
||||
if i == 0 {
|
||||
session, _ = r["sessionId"].(string)
|
||||
turn, _ = r["turnId"].(string)
|
||||
}
|
||||
if r["sessionId"] != session || r["turnId"] != turn || session == "" || !strings.HasPrefix(turn, "turn_") {
|
||||
t.Fatalf("line %d envelope = %v/%v, want one session and one turn_ id", i, r["sessionId"], r["turnId"])
|
||||
}
|
||||
if !vocab[r.kind()] {
|
||||
t.Fatalf("line %d has kind %q, which 1.x never emitted", i, r.kind())
|
||||
}
|
||||
}
|
||||
for _, r := range v1 {
|
||||
if r.kind() == "" {
|
||||
continue
|
||||
}
|
||||
if !slices.ContainsFunc(events, func(g streamRecord) bool { return g.kind() == r.kind() }) {
|
||||
t.Fatalf("1.x emitted %q for this run; the stream here has none", r.kind())
|
||||
}
|
||||
}
|
||||
wantStatus := map[string]string{"turn_status": "queued", "turn_done": "completed", "turn_started": "in_progress", "tool_started": "in_progress"}
|
||||
for _, r := range events {
|
||||
if want, ok := wantStatus[r.kind()]; ok && r["status"] != want {
|
||||
t.Fatalf("%s status = %v, want %s", r.kind(), r["status"], want)
|
||||
}
|
||||
if r.kind() == "user_message" && r["text"] != "READ note.txt" {
|
||||
t.Fatalf("user_message text = %v, want the prompt", r["text"])
|
||||
}
|
||||
}
|
||||
if session != got[len(got)-1]["session_id"] {
|
||||
t.Fatalf("envelope sessionId %q differs from the result's session_id %v", session, got[len(got)-1]["session_id"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunEventsJSONLKeeps1xLifecycle(t *testing.T) {
|
||||
v1 := readV1Sample(t, "events-jsonl-write-file.jsonl")
|
||||
runToolCallFixture(t, "read_file", `{"path":"note.txt"}`)
|
||||
got := runCLIStream(t, "run", "--events-jsonl", "READ note.txt")
|
||||
|
||||
if want, have := lifecycleSkeleton(v1), lifecycleSkeleton(got); !slices.Equal(want, have) {
|
||||
t.Fatalf("lifecycle records = %v, want 1.x's %v", have, want)
|
||||
}
|
||||
vocab := streamVocabulary()
|
||||
for i, r := range got {
|
||||
if r["sequence"] != float64(i+1) || r["schema_version"] != float64(machineSchemaVersion) {
|
||||
t.Fatalf("line %d envelope = %v", i, r)
|
||||
}
|
||||
if !vocab[r.kind()] {
|
||||
t.Fatalf("line %d has kind %q, which 1.x never emitted", i, r.kind())
|
||||
}
|
||||
if r.kind() == "user_message" && len(r) != 3 {
|
||||
t.Fatalf("events-jsonl user_message must stay content-free: %v", r)
|
||||
}
|
||||
}
|
||||
if last := got[len(got)-1]; last.kind() != "run_done" || last["ok"] != true || last["num_turns"] != float64(1) {
|
||||
t.Fatalf("last line = %v, want run_done ok with num_turns 1", last)
|
||||
}
|
||||
}
|
||||
|
||||
// A call the permission gate refuses never ran, so it has a result and no
|
||||
// tool_started.
|
||||
func TestRunStreamJSONReportsNoStartForARefusedCall(t *testing.T) {
|
||||
dir := runToolCallFixture(t, "write_file", `{"path":"marker.txt","content":"x"}`)
|
||||
got := runCLIStream(t, "run", "--output-format", "stream-json", "WRITE marker.txt")
|
||||
if _, err := os.Stat(filepath.Join(dir, "marker.txt")); err == nil {
|
||||
t.Fatal("the default headless mode wrote the file; this case needs a refused call")
|
||||
}
|
||||
skeleton := lifecycleSkeleton(got)
|
||||
if slices.Contains(skeleton, "tool_started") || !slices.Contains(skeleton, "tool_result") {
|
||||
t.Fatalf("refused call lifecycle = %v, want a tool_result and no tool_started", skeleton)
|
||||
}
|
||||
}
|
||||
|
||||
// A turn the provider fails still ran: 1.x counted it and closed it with
|
||||
// turn_done, status failed.
|
||||
func TestRunStreamJSONClosesAFailedTurn(t *testing.T) {
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
_, _ = io.Copy(io.Discard, r.Body)
|
||||
http.Error(w, `{"error":{"message":"bad request"}}`, http.StatusBadRequest)
|
||||
}))
|
||||
t.Cleanup(srv.Close)
|
||||
isolateCLIConfigHome(t)
|
||||
home := testenv.TempDir(t)
|
||||
t.Setenv("REASONIX_HOME", home)
|
||||
t.Setenv("RUN_FLAGS_FAKE_KEY", "k")
|
||||
writeTestFile(t, filepath.Join(home, "config.toml"), fmt.Sprintf(`default_model = "fake"
|
||||
|
||||
[[providers]]
|
||||
name = "fake"
|
||||
kind = "openai"
|
||||
base_url = %q
|
||||
model = "fake-model"
|
||||
api_key_env = "RUN_FLAGS_FAKE_KEY"
|
||||
`, srv.URL), 0o644)
|
||||
t.Chdir(testenv.TempDir(t))
|
||||
var code int
|
||||
stdout, stderr := captureDrainedOutput(t, func() { code = Run([]string{"run", "--output-format", "stream-json", "hi"}, "test") })
|
||||
if code != 1 {
|
||||
t.Fatalf("exit %d, want 1\nstdout:\n%s\nstderr:\n%s", code, stdout, stderr)
|
||||
}
|
||||
got := readStreamRecords(t, stdout)
|
||||
done := got[len(got)-2]
|
||||
if done.kind() != "turn_done" || done["status"] != "failed" {
|
||||
t.Fatalf("second-to-last line = %v, want turn_done failed", done)
|
||||
}
|
||||
if result := got[len(got)-1]; result["is_error"] != true || result["num_turns"] != float64(1) {
|
||||
t.Fatalf("result = %v, want is_error with num_turns 1", result)
|
||||
}
|
||||
}
|
||||
|
||||
// captureDrainedOutput reads both streams while fn runs: a stream-json run
|
||||
// outgrows a Windows pipe buffer, and an unread pipe blocks the writer.
|
||||
func captureDrainedOutput(t *testing.T, fn func()) (stdout, stderr string) {
|
||||
t.Helper()
|
||||
drain := func(target **os.File) func() string {
|
||||
r, w, err := os.Pipe()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
*target = w
|
||||
done := make(chan string, 1)
|
||||
go func() {
|
||||
data, _ := io.ReadAll(r)
|
||||
done <- string(data)
|
||||
}()
|
||||
return func() string { _ = w.Close(); return <-done }
|
||||
}
|
||||
oldOut, oldErr := os.Stdout, os.Stderr
|
||||
defer func() { os.Stdout, os.Stderr = oldOut, oldErr }()
|
||||
outDone := drain(&os.Stdout)
|
||||
errDone := drain(&os.Stderr)
|
||||
fn()
|
||||
return outDone(), errDone()
|
||||
}
|
||||
@@ -0,0 +1,151 @@
|
||||
package cli
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
|
||||
"reasonix/internal/contract/event"
|
||||
"reasonix/internal/contract/eventwire"
|
||||
)
|
||||
|
||||
// Headless machine streams keep 1.x's line contract (per-line envelope plus
|
||||
// turn_status, user_message, tool_started, turn_done). The synchronous
|
||||
// controller path publishes no turn lifecycle, so this transport supplies it.
|
||||
|
||||
// runStreamKinds is the vocabulary both machine streams carry: the kinds 1.x
|
||||
// emitted. The rest describe host-internal state (leases, inbox, graph, tabs).
|
||||
var runStreamKinds = map[event.Kind]bool{
|
||||
event.TurnStarted: true, event.Reasoning: true, event.Text: true, event.Message: true,
|
||||
event.ToolDispatch: true, event.ToolResult: true, event.ToolProgress: true, event.Usage: true,
|
||||
event.Notice: true, event.Phase: true, event.TurnPhase: true, event.ApprovalRequest: true,
|
||||
event.AskRequest: true, event.TurnDone: true, event.CompactionStarted: true,
|
||||
event.CompactionDone: true, event.MCPSurfaceReady: true, event.Retrying: true, event.Steer: true,
|
||||
event.GuardianAssessment: true, event.ExtensionSurface: true, event.ExtensionStatus: true,
|
||||
event.StreamAttempt: true, event.ContextMaintenanceEvent: true, event.WorkspaceChanged: true,
|
||||
event.CompletionSummary: true,
|
||||
}
|
||||
|
||||
const (
|
||||
runTurnQueued = "queued"
|
||||
runTurnInProgress = "in_progress"
|
||||
runTurnCompleted = "completed"
|
||||
runTurnFailed = "failed"
|
||||
runTurnInterrupted = "interrupted"
|
||||
)
|
||||
|
||||
// runTurnEnvelope is the one top-level turn a headless run drives.
|
||||
type runTurnEnvelope struct {
|
||||
sessionID string
|
||||
turnID string
|
||||
prompt string
|
||||
begun bool
|
||||
announced bool
|
||||
}
|
||||
|
||||
type streamJSONLine struct {
|
||||
eventwire.Event
|
||||
SessionID string `json:"sessionId,omitempty"`
|
||||
TurnID string `json:"turnId,omitempty"`
|
||||
Seq uint64 `json:"seq"`
|
||||
Status string `json:"status,omitempty"`
|
||||
}
|
||||
|
||||
func newRunTurnID() string {
|
||||
var b [16]byte
|
||||
_, _ = rand.Read(b[:])
|
||||
return "turn_" + hex.EncodeToString(b[:])
|
||||
}
|
||||
|
||||
func runTurnTerminalStatus(runErr error, completion runCompletion) string {
|
||||
switch {
|
||||
case !completion.isError:
|
||||
return runTurnCompleted
|
||||
case errors.Is(runErr, context.Canceled):
|
||||
return runTurnInterrupted
|
||||
default:
|
||||
return runTurnFailed
|
||||
}
|
||||
}
|
||||
|
||||
// BeginTurn opens the turn the run is about to submit and reports it queued.
|
||||
func (s *runOutputSink) BeginTurn(sessionID, prompt string) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.turn = runTurnEnvelope{sessionID: sessionID, turnID: newRunTurnID(), prompt: prompt, begun: true}
|
||||
s.writeLifecycle(event.Event{}, "turn_status", runTurnQueued)
|
||||
}
|
||||
|
||||
// RecordTurnCompletion counts a turn the controller admitted and finished.
|
||||
func (s *runOutputSink) RecordTurnCompletion() {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.turns++
|
||||
}
|
||||
|
||||
// writeStreamEvent writes one kernel event, bracketed by the lifecycle records
|
||||
// 1.x placed around it.
|
||||
func (s *runOutputSink) writeStreamEvent(e event.Event) {
|
||||
if !runStreamKinds[e.Kind] || s.err != nil {
|
||||
return
|
||||
}
|
||||
if e.Kind == event.ToolResult && e.Tool.Executed {
|
||||
s.writeLifecycle(event.Event{Kind: event.ToolDispatch, Tool: event.Tool{
|
||||
ID: e.Tool.ID, Name: e.Tool.Name, ReadOnly: e.Tool.ReadOnly, StartedAt: e.Tool.StartedAt,
|
||||
ParentID: e.Tool.ParentID, Issuer: e.Tool.Issuer,
|
||||
}}, "tool_started", s.inTurnStatus())
|
||||
}
|
||||
if s.format == runOutputStreamJSON {
|
||||
s.sequence++
|
||||
s.err = s.encoder.Encode(s.streamLine(eventwire.ToWire(e), s.inTurnStatus()))
|
||||
} else {
|
||||
s.sequence++
|
||||
s.err = s.encoder.Encode(s.machineEventRecordFor(e, s.sequence))
|
||||
}
|
||||
if e.Kind == event.TurnStarted && s.turn.begun && !s.turn.announced {
|
||||
s.turn.announced = true
|
||||
s.writeLifecycle(event.Event{Text: s.turn.prompt}, "user_message", s.inTurnStatus())
|
||||
}
|
||||
}
|
||||
|
||||
// finishTurn closes a begun turn with its terminal status.
|
||||
func (s *runOutputSink) finishTurn(runErr error, completion runCompletion) {
|
||||
if !s.turn.begun || s.err != nil {
|
||||
return
|
||||
}
|
||||
done := event.Event{Kind: event.TurnDone, Err: runErr, Outcome: completion.outcome, Cancelled: errors.Is(runErr, context.Canceled)}
|
||||
s.writeLifecycle(done, "turn_done", runTurnTerminalStatus(runErr, completion))
|
||||
}
|
||||
|
||||
// writeLifecycle writes a record this transport supplies under kind. The
|
||||
// event gives its payload; for events-jsonl only content-free fields survive.
|
||||
func (s *runOutputSink) writeLifecycle(e event.Event, kind, status string) {
|
||||
if s.err != nil || (s.format != runOutputStreamJSON && s.format != runOutputEventsJSONL) {
|
||||
return
|
||||
}
|
||||
s.sequence++
|
||||
if s.format == runOutputStreamJSON {
|
||||
w := eventwire.ToWire(e)
|
||||
w.Kind = kind
|
||||
s.err = s.encoder.Encode(s.streamLine(w, status))
|
||||
return
|
||||
}
|
||||
record := machineEventRecord{SchemaVersion: machineSchemaVersion, Sequence: s.sequence, Kind: kind}
|
||||
if e.Kind == event.ToolDispatch || e.Kind == event.TurnDone {
|
||||
record = s.machineEventRecordFor(e, s.sequence)
|
||||
record.Kind = kind
|
||||
}
|
||||
s.err = s.encoder.Encode(record)
|
||||
}
|
||||
|
||||
func (s *runOutputSink) streamLine(w eventwire.Event, status string) streamJSONLine {
|
||||
return streamJSONLine{Event: w, SessionID: s.turn.sessionID, TurnID: s.turn.turnID, Seq: s.sequence, Status: status}
|
||||
}
|
||||
|
||||
func (s *runOutputSink) inTurnStatus() string {
|
||||
if !s.turn.begun {
|
||||
return ""
|
||||
}
|
||||
return runTurnInProgress
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
{"schema_version":1,"sequence":1,"kind":"turn_status"}
|
||||
{"schema_version":1,"sequence":2,"kind":"turn_started"}
|
||||
{"schema_version":1,"sequence":3,"kind":"turn_phase"}
|
||||
{"schema_version":1,"sequence":4,"kind":"user_message"}
|
||||
{"schema_version":1,"sequence":5,"kind":"stream_attempt"}
|
||||
{"schema_version":1,"sequence":6,"kind":"tool_dispatch","tool_id":"tool_1","tool_name":"tool_name_1"}
|
||||
{"schema_version":1,"sequence":7,"kind":"usage","usage":{"input_tokens":100,"output_tokens":10,"cache_hit_tokens":0,"cache_miss_tokens":100}}
|
||||
{"schema_version":1,"sequence":8,"kind":"stream_attempt"}
|
||||
{"schema_version":1,"sequence":9,"kind":"turn_phase"}
|
||||
{"schema_version":1,"sequence":10,"kind":"tool_dispatch","tool_id":"tool_1","tool_name":"tool_name_1"}
|
||||
{"schema_version":1,"sequence":11,"kind":"tool_started"}
|
||||
{"schema_version":1,"sequence":12,"kind":"tool_result","tool_id":"tool_1","tool_name":"tool_name_1","tool_duration_ms":80}
|
||||
{"schema_version":1,"sequence":13,"kind":"turn_phase"}
|
||||
{"schema_version":1,"sequence":14,"kind":"stream_attempt"}
|
||||
{"schema_version":1,"sequence":15,"kind":"text"}
|
||||
{"schema_version":1,"sequence":16,"kind":"usage","usage":{"input_tokens":100,"output_tokens":10,"cache_hit_tokens":0,"cache_miss_tokens":100}}
|
||||
{"schema_version":1,"sequence":17,"kind":"message"}
|
||||
{"schema_version":1,"sequence":18,"kind":"stream_attempt"}
|
||||
{"schema_version":1,"sequence":19,"kind":"turn_done"}
|
||||
{"schema_version":1,"sequence":20,"kind":"run_done","ok":true,"duration_ms":401,"num_turns":1,"usage":{"input_tokens":200,"output_tokens":20,"cache_hit_tokens":0,"cache_miss_tokens":200}}
|
||||
@@ -0,0 +1,20 @@
|
||||
{"kind":"turn_status","sessionId":"dba52b11e381ce9892ff6dfcc5d1d0fe","turnId":"turn_a5e5da456c8821227fca726e488eb5c0","seq":1,"status":"queued"}
|
||||
{"kind":"turn_started","sessionId":"dba52b11e381ce9892ff6dfcc5d1d0fe","turnId":"turn_a5e5da456c8821227fca726e488eb5c0","seq":2,"status":"in_progress"}
|
||||
{"kind":"turn_phase","sessionId":"dba52b11e381ce9892ff6dfcc5d1d0fe","turnId":"turn_a5e5da456c8821227fca726e488eb5c0","seq":3,"status":"in_progress","text":"working","phase":"working"}
|
||||
{"kind":"user_message","messageId":"01M3NQNJCZMPJWDD7597QAPTG6","sessionId":"dba52b11e381ce9892ff6dfcc5d1d0fe","turnId":"turn_a5e5da456c8821227fca726e488eb5c0","seq":4,"status":"in_progress","text":"WRITEFILE"}
|
||||
{"kind":"stream_attempt","messageId":"01M3NQNJCZAJJ6WDZ86DH246Z3","attemptId":"01M3NQNJCZAJJ6WDZ86DH246Z3","sessionId":"dba52b11e381ce9892ff6dfcc5d1d0fe","turnId":"turn_a5e5da456c8821227fca726e488eb5c0","seq":5,"status":"in_progress","streamAttempt":{"id":"01M3NQNJCZAJJ6WDZ86DH246Z3","action":"begin","attempt":1,"max":4}}
|
||||
{"kind":"tool_dispatch","messageId":"01M3NQNJCZAJJ6WDZ86DH246Z3","attemptId":"01M3NQNJCZAJJ6WDZ86DH246Z3","sessionId":"dba52b11e381ce9892ff6dfcc5d1d0fe","turnId":"turn_a5e5da456c8821227fca726e488eb5c0","seq":6,"status":"in_progress","tool":{"id":"call_1","name":"write_file","todos":[],"readOnly":false,"partial":true,"attemptId":"01M3NQNJCZAJJ6WDZ86DH246Z3"}}
|
||||
{"kind":"usage","sessionId":"dba52b11e381ce9892ff6dfcc5d1d0fe","turnId":"turn_a5e5da456c8821227fca726e488eb5c0","seq":7,"status":"in_progress","usage":{"promptTokens":100,"completionTokens":10,"totalTokens":110,"cacheHitTokens":0,"cacheMissTokens":100,"source":"executor","cacheDiagnostics":{"prefixHash":"9f253dbb5bb109ca","prefixChanged":false,"systemHash":"adc39bc9d14f8e34","toolsHash":"69878e44e84d9a93","logRewriteVersion":0,"toolSchemaTokens":3409,"cacheMissTokens":100,"cacheHitTokens":0,"sessionContext":{"version":1,"digest":"9cf8c9c180ecedda47e13c6569be9d4ab55361b973cc5b120b1e9696d93dffd5","targetRole":"executor","reasons":["first_seen"],"environment":{"digest":"cb1b8caa6059dc8bfc7d78d00e673d886d95f4511bdf98ffd65dfbad906b5c36","chars":421},"workspace":{"digest":"3181a58accb19118efe53f270e7398a336bf46dc92d18361fbaea8c4462e8a54","chars":155},"backgroundMemory":{},"skillsCatalog":{"digest":"6ee6f4ddda6b8bf82bf96b05dc410aba3e3cac11aa20c3cfc433f14c57473a15","chars":879}}},"sessionCacheHitTokens":0,"sessionCacheMissTokens":100,"contextPromptTokens":100,"contextCompletionTokens":10,"contextCacheMissTokens":100,"costQuote":{"original":{"amount":"","currency":""},"estimated":true,"costComplete":false,"displayComplete":false,"complete":false,"displayStatus":"unavailable","modelRef":"mock/mock-model","usageSource":"executor","incompleteReason":"no_price"},"displayStatus":"unavailable"}}
|
||||
{"kind":"stream_attempt","messageId":"01M3NQNJCZAJJ6WDZ86DH246Z3","attemptId":"01M3NQNJCZAJJ6WDZ86DH246Z3","sessionId":"dba52b11e381ce9892ff6dfcc5d1d0fe","turnId":"turn_a5e5da456c8821227fca726e488eb5c0","seq":8,"status":"in_progress","streamAttempt":{"id":"01M3NQNJCZAJJ6WDZ86DH246Z3","action":"commit","attempt":1,"max":4}}
|
||||
{"kind":"turn_phase","sessionId":"dba52b11e381ce9892ff6dfcc5d1d0fe","turnId":"turn_a5e5da456c8821227fca726e488eb5c0","seq":9,"status":"in_progress","text":"checking","phase":"checking"}
|
||||
{"kind":"tool_dispatch","messageId":"01M3NQNJCZAJJ6WDZ86DH246Z3","sessionId":"dba52b11e381ce9892ff6dfcc5d1d0fe","turnId":"turn_a5e5da456c8821227fca726e488eb5c0","seq":10,"status":"in_progress","tool":{"runState":"pending","id":"call_1","name":"write_file","args":"{\"path\": \"hello.txt\", \"content\": \"hi\\n\"}","todos":[],"readOnly":false,"diff":"--- a//work/hello.txt\n+++ b//work/hello.txt\n@@ -0,0 +1 @@\n+hi\n","added":1}}
|
||||
{"kind":"tool_started","sessionId":"dba52b11e381ce9892ff6dfcc5d1d0fe","turnId":"turn_a5e5da456c8821227fca726e488eb5c0","seq":11,"status":"in_progress","tool":{"runState":"started","id":"call_1","name":"write_file","todos":[],"readOnly":false,"attemptId":"8eb7f57e64a0687cfe17a78af6f23d18"}}
|
||||
{"kind":"tool_result","messageId":"01M3NQNJCZAJJ6WDZ86DH246Z3","sessionId":"dba52b11e381ce9892ff6dfcc5d1d0fe","turnId":"turn_a5e5da456c8821227fca726e488eb5c0","seq":12,"status":"in_progress","tool":{"runState":"completed","id":"call_1","name":"write_file","args":"{\"path\": \"hello.txt\", \"content\": \"hi\\n\"}","todos":[],"output":"wrote 3 bytes to /work/hello.txt","readOnly":false,"durationMs":86,"startedAt":1790657087949,"endedAt":1790657088035}}
|
||||
{"kind":"turn_phase","sessionId":"dba52b11e381ce9892ff6dfcc5d1d0fe","turnId":"turn_a5e5da456c8821227fca726e488eb5c0","seq":13,"status":"in_progress","text":"working","phase":"working"}
|
||||
{"kind":"stream_attempt","messageId":"01M3NQNJH5C21RTNN04H8YDNC2","attemptId":"01M3NQNJH5C21RTNN04H8YDNC2","sessionId":"dba52b11e381ce9892ff6dfcc5d1d0fe","turnId":"turn_a5e5da456c8821227fca726e488eb5c0","seq":14,"status":"in_progress","streamAttempt":{"id":"01M3NQNJH5C21RTNN04H8YDNC2","action":"begin","attempt":1,"max":4}}
|
||||
{"kind":"text","messageId":"01M3NQNJH5C21RTNN04H8YDNC2","attemptId":"01M3NQNJH5C21RTNN04H8YDNC2","sessionId":"dba52b11e381ce9892ff6dfcc5d1d0fe","turnId":"turn_a5e5da456c8821227fca726e488eb5c0","seq":15,"status":"in_progress","text":"MOCK-ANSWER after-tool: wrote 3 bytes to /work/hello.txt"}
|
||||
{"kind":"usage","sessionId":"dba52b11e381ce9892ff6dfcc5d1d0fe","turnId":"turn_a5e5da456c8821227fca726e488eb5c0","seq":16,"status":"in_progress","usage":{"promptTokens":100,"completionTokens":10,"totalTokens":110,"cacheHitTokens":0,"cacheMissTokens":100,"source":"executor","cacheDiagnostics":{"prefixHash":"9f253dbb5bb109ca","prefixChanged":false,"systemHash":"adc39bc9d14f8e34","toolsHash":"69878e44e84d9a93","logRewriteVersion":1,"toolSchemaTokens":3409,"cacheMissTokens":100,"cacheHitTokens":0,"sessionContext":{"version":1,"digest":"9cf8c9c180ecedda47e13c6569be9d4ab55361b973cc5b120b1e9696d93dffd5","targetRole":"executor","reasons":["first_seen"],"environment":{"digest":"cb1b8caa6059dc8bfc7d78d00e673d886d95f4511bdf98ffd65dfbad906b5c36","chars":421},"workspace":{"digest":"3181a58accb19118efe53f270e7398a336bf46dc92d18361fbaea8c4462e8a54","chars":155},"backgroundMemory":{},"skillsCatalog":{"digest":"6ee6f4ddda6b8bf82bf96b05dc410aba3e3cac11aa20c3cfc433f14c57473a15","chars":879}}},"sessionCacheHitTokens":0,"sessionCacheMissTokens":200,"contextPromptTokens":100,"contextCompletionTokens":10,"contextCacheMissTokens":100,"costQuote":{"original":{"amount":"","currency":""},"estimated":true,"costComplete":false,"displayComplete":false,"complete":false,"displayStatus":"unavailable","modelRef":"mock/mock-model","usageSource":"executor","incompleteReason":"no_price"},"displayStatus":"unavailable"}}
|
||||
{"kind":"message","messageId":"01M3NQNJH5C21RTNN04H8YDNC2","attemptId":"01M3NQNJH5C21RTNN04H8YDNC2","sessionId":"dba52b11e381ce9892ff6dfcc5d1d0fe","turnId":"turn_a5e5da456c8821227fca726e488eb5c0","seq":17,"status":"in_progress","text":"MOCK-ANSWER after-tool: wrote 3 bytes to /work/hello.txt"}
|
||||
{"kind":"stream_attempt","messageId":"01M3NQNJH5C21RTNN04H8YDNC2","attemptId":"01M3NQNJH5C21RTNN04H8YDNC2","sessionId":"dba52b11e381ce9892ff6dfcc5d1d0fe","turnId":"turn_a5e5da456c8821227fca726e488eb5c0","seq":18,"status":"in_progress","streamAttempt":{"id":"01M3NQNJH5C21RTNN04H8YDNC2","action":"commit","attempt":1,"max":4}}
|
||||
{"kind":"turn_done","sessionId":"dba52b11e381ce9892ff6dfcc5d1d0fe","turnId":"turn_a5e5da456c8821227fca726e488eb5c0","seq":19,"status":"completed"}
|
||||
{"type":"result","subtype":"success","is_error":false,"duration_ms":427,"num_turns":1,"result":"MOCK-ANSWER after-tool: wrote 3 bytes to /work/hello.txt","cost_complete":false,"display_complete":false,"display_status":"unavailable","cost_quote":{"original":{"amount":"0","currency":""},"estimated":true,"costComplete":false,"displayComplete":false,"complete":false,"displayStatus":"unavailable","incompleteReason":"no_price"},"usage":{"input_tokens":200,"output_tokens":20,"cache_read_input_tokens":0,"cache_creation_input_tokens":200}}
|
||||
@@ -321,7 +321,7 @@ func (a *Agent) executeBatch(ctx context.Context, turn *turnRuntime, calls []pro
|
||||
Bound: o.bound,
|
||||
DurationMs: durations[i],
|
||||
Execution: toEventShellExecution(o.execution, durations[i]),
|
||||
Issuer: event.IssuedByModel,
|
||||
Issuer: event.IssuedByModel, Executed: o.executed,
|
||||
}
|
||||
if startedAt[i] > 0 {
|
||||
tr.StartedAt = startedAt[i]
|
||||
|
||||
@@ -35,6 +35,7 @@ func (n *nestedInvoker) Invoke(ctx context.Context, name string, args json.RawMe
|
||||
ID: call.ID, Name: name, Args: call.Arguments, Output: out.output, Err: out.errMsg,
|
||||
RefusalCode: out.refusalCode, Bound: out.bound, DurationMs: elapsed,
|
||||
StartedAt: started.UnixMilli(), EndedAt: started.UnixMilli() + elapsed, Issuer: event.IssuedByModel,
|
||||
Executed: out.executed,
|
||||
}})
|
||||
if out.provenance.External() && !n.parent.nestedOrigin.External() {
|
||||
n.parent.nestedOrigin = out.provenance
|
||||
|
||||
@@ -137,13 +137,21 @@ func (r *Recorder) Emit(e event.Event) {
|
||||
} else if r != nil && r.writer != nil && e.Kind == event.GuardianAssessment && e.Guardian.Usage != nil {
|
||||
r.recordProviderUsage(e.ModelRef, e.Guardian.Usage, nil, "")
|
||||
} else if r != nil && r.writer != nil && e.Kind == event.TurnDone {
|
||||
r.RecordTurnCompletion()
|
||||
r.recordTurn()
|
||||
}
|
||||
}
|
||||
|
||||
// RecordTurnCompletion records synchronous controller runs that deliberately do
|
||||
// not emit TurnDone into the UI event stream.
|
||||
// not emit TurnDone into the UI event stream, and passes the signal inward.
|
||||
func (r *Recorder) RecordTurnCompletion() {
|
||||
if r == nil {
|
||||
return
|
||||
}
|
||||
r.recordTurn()
|
||||
event.RecordTurnCompletion(r.inner)
|
||||
}
|
||||
|
||||
func (r *Recorder) recordTurn() {
|
||||
if r == nil || r.dispatcher == nil {
|
||||
return
|
||||
}
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
package stats
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"reasonix/internal/base/testenv"
|
||||
"reasonix/internal/contract/event"
|
||||
)
|
||||
|
||||
type turnCompletionSpy struct {
|
||||
spySink
|
||||
completions int
|
||||
}
|
||||
|
||||
func (s *turnCompletionSpy) RecordTurnCompletion() { s.completions++ }
|
||||
|
||||
// A synchronous run reports its turn through RecordTurnCompletion; the
|
||||
// recorder must pass it to the frontend sink it wraps, once, and must not
|
||||
// turn a TurnDone event into a second completion.
|
||||
func TestRecorderForwardsTurnCompletionOnce(t *testing.T) {
|
||||
inner := &turnCompletionSpy{}
|
||||
r := NewRecorder(inner, testenv.TempDir(t), "cli")
|
||||
event.RecordTurnCompletion(r)
|
||||
r.Emit(turnEvent())
|
||||
flushRecorder(t, r)
|
||||
if inner.completions != 1 {
|
||||
t.Fatalf("inner saw %d turn completions, want 1", inner.completions)
|
||||
}
|
||||
if len(inner.events) != 1 || inner.events[0].Kind != event.TurnDone {
|
||||
t.Fatalf("inner events = %+v, want the one TurnDone", inner.events)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user