fix: unify runner egress and AOP persistence

agent/aop_emit.go

@@ -102,6 +102,7 @@ func (e *aopEmitter) sessionStart(model string) {
 	event := &aop.Event{Payload: &aop.Event_SessionStarted{SessionStarted: &aop.SessionStarted{
 		Model: model, ParentSessionId: e.parentSessionID, ParentToolCallId: e.parentToolCallID,
 	}}}
+	_ = types.SetSessionHistory(event, &types.SessionHistory{Mode: types.SessionHistory_MODE_INHERIT})
 	if e.delegation != nil {
 		e.emitWithExt(event, e.delegation)
 		return

core/config/options.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"runtime"
 	"strings"
 
 	"github.com/chainreactors/aiscan/skills"
@@ -241,6 +242,12 @@ func ResolvePrompt(value string) (string, error) {
 		return prompt, nil
 	}
 	if err != nil {
+		// Natural-language prompts frequently contain punctuation that is not
+		// legal in a Windows filename (for example `host:port`). An invalid
+		// filename is evidence that this is text, not a prompt-file request.
+		if runtime.GOOS == "windows" && strings.ContainsAny(prompt, `<>:"|?*`) {
+			return prompt, nil
+		}
 		return "", fmt.Errorf("stat prompt file %s: %w", prompt, err)
 	}
 	if !info.Mode().IsRegular() {

core/config/options_test.go

@@ -33,6 +33,17 @@ func TestResolvePromptKeepsMissingPathAsNaturalLanguage(t *testing.T) {
 	}
 }
 
+func TestResolvePromptKeepsWindowsInvalidFilenameAsNaturalLanguage(t *testing.T) {
+	prompt := "check host:port and report the result"
+	got, err := ResolvePrompt(prompt)
+	if err != nil {
+		t.Fatalf("ResolvePrompt() error = %v", err)
+	}
+	if got != prompt {
+		t.Fatalf("ResolvePrompt() = %q, want %q", got, prompt)
+	}
+}
+
 func TestResolveTaskLoadsPromptFileAndAppendsInputs(t *testing.T) {
 	path := filepath.Join(t.TempDir(), "task.md")
 	if err := os.WriteFile(path, []byte("inspect the exposed services"), 0o600); err != nil {

core/output/jsonl.go

@@ -2,6 +2,7 @@ package output
 
 import (
 	"bufio"
+	"bytes"
 	"fmt"
 	"io"
 	"os"
@@ -15,7 +16,8 @@ import (
 )
 
 // ScanJSONL decodes the canonical append-only AOP event stream one line at a
-// time. Empty and non-JSON lines are ignored; malformed AOP event lines fail.
+// time. Blank lines are allowed; every non-blank line must be a complete AOP
+// event with a session and payload.
 func ScanJSONL(path string, visit func(*aop.Event) error) error {
 	file, err := os.Open(path)
 	if err != nil {
@@ -25,16 +27,19 @@ func ScanJSONL(path string, visit func(*aop.Event) error) error {
 	scanner := bufio.NewScanner(file)
 	scanner.Buffer(make([]byte, 0, 256*1024), 64*1024*1024)
 	for scanner.Scan() {
-		line := scanner.Bytes()
-		if len(line) == 0 || line[0] != '{' {
+		line := bytes.TrimSpace(scanner.Bytes())
+		if len(line) == 0 {
 			continue
 		}
+		if line[0] != '{' {
+			return fmt.Errorf("AOP JSONL contains a non-event line")
+		}
 		event := new(aop.Event)
 		if err := protojson.Unmarshal(line, event); err != nil {
 			return fmt.Errorf("decode AOP JSONL event: %w", err)
 		}
-		if event.SessionId == "" || event.Payload == nil {
-			continue
+		if event.Id == "" || event.SessionId == "" || event.Payload == nil {
+			return fmt.Errorf("AOP JSONL event is missing id, session_id or payload")
 		}
 		if visit != nil {
 			if err := visit(event); err != nil {
@@ -57,30 +62,23 @@ func ReadJSONL(path string) ([]*aop.Event, error) {
 	return events, err
 }
 
-// DefaultRecorderMaxBytes caps one session's JSONL file. Legitimate sessions
-// stay in the megabytes; the cap is a last-line defense so a runaway event
-// source (a retry loop once emitted turn lifecycle pairs at microsecond
-// cadence and wrote 182GB) cannot fill the disk. It is a capacity guard, not
-// content policy: no event inspection or deduplication happens here.
-const DefaultRecorderMaxBytes int64 = 2 << 30 // 2 GiB
-
 // JSONLRecorder is the single append-only subscriber for persisted AOP events.
 type JSONLRecorder struct {
-	mu       sync.Mutex
-	file     *os.File
-	path     string
-	unsub    func()
-	err      error
-	maxBytes int64
-	size     int64
-	limited  bool
+	mu   sync.Mutex
+	file *os.File
+	path string
+	// seen is scoped to this recorder. Event IDs are unique within a session,
+	// not necessarily across independent sessions, so the key includes both.
+	seen  map[string]struct{}
+	unsub func()
+	err   error
 }
 
 func NewJSONLRecorder(bus *eventbus.Bus[*aop.Event], path string) (*JSONLRecorder, error) {
 	if bus == nil {
 		return nil, fmt.Errorf("AOP event bus is required")
 	}
-	recorder := &JSONLRecorder{maxBytes: DefaultRecorderMaxBytes}
+	recorder := &JSONLRecorder{seen: make(map[string]struct{})}
 	if err := recorder.Switch(path); err != nil {
 		return nil, err
 	}
@@ -126,31 +124,40 @@ func (r *JSONLRecorder) Switch(path string) error {
 	if err != nil {
 		return err
 	}
-	// The file is opened in append mode, so a resumed session starts with its
-	// existing size already counted against the cap.
-	var size int64
-	if info, statErr := file.Stat(); statErr == nil {
-		size = info.Size()
-	}
 	r.mu.Lock()
+	defer r.mu.Unlock()
+	seen, err := loadJSONLIDs(clean)
+	if err != nil {
+		_ = file.Close()
+		return err
+	}
 	old := r.file
 	r.file = file
 	r.path = clean
-	r.size = size
-	r.limited = r.maxBytes > 0 && size >= r.maxBytes
-	r.mu.Unlock()
+	r.seen = seen
 	if old != nil {
 		if err := old.Close(); err != nil {
-			r.mu.Lock()
 			if r.err == nil {
 				r.err = err
 			}
-			r.mu.Unlock()
 		}
 	}
 	return nil
 }
 
+func loadJSONLIDs(path string) (map[string]struct{}, error) {
+	seen := make(map[string]struct{})
+	if err := ScanJSONL(path, func(event *aop.Event) error {
+		if event.Id != "" {
+			seen[event.SessionId+"\x00"+event.Id] = struct{}{}
+		}
+		return nil
+	}); err != nil {
+		return nil, err
+	}
+	return seen, nil
+}
+
 func (r *JSONLRecorder) Path() string {
 	if r == nil {
 		return ""
@@ -164,13 +171,8 @@ func (r *JSONLRecorder) Write(event *aop.Event) error {
 	if r == nil || event == nil {
 		return nil
 	}
-	// Fast path once the cap has tripped: skip the marshal so an ongoing
-	// event storm costs almost nothing per dropped event.
-	r.mu.Lock()
-	limited := r.limited
-	r.mu.Unlock()
-	if limited {
-		return nil
+	if event.Id == "" || event.SessionId == "" || event.Payload == nil {
+		return fmt.Errorf("AOP JSONL event requires id, session_id and payload")
 	}
 	line, err := (protojson.MarshalOptions{UseProtoNames: true}).Marshal(event)
 	if err != nil {
@@ -182,28 +184,22 @@ func (r *JSONLRecorder) Write(event *aop.Event) error {
 	if r.file == nil {
 		return io.ErrClosedPipe
 	}
-	if r.limited {
+	key := event.SessionId + "\x00" + event.Id
+	if _, exists := r.seen[key]; exists {
 		return nil
 	}
-	if r.maxBytes > 0 && r.size+int64(len(line)) > r.maxBytes {
-		r.limited = true
-		// Leave one non-JSON marker line as durable evidence of the
-		// truncation; ScanJSONL skips lines that do not start with '{', so
-		// the file stays readable and resumable.
-		marker := fmt.Sprintf("# aiscan: JSONL size limit reached (limit=%d bytes); subsequent events are dropped\n", r.maxBytes)
-		_, _ = r.file.Write([]byte(marker))
-		// The bus subscription stores the first Write error, so this
-		// surfaces once through Close instead of once per dropped event.
-		return fmt.Errorf("AOP JSONL %s reached the %d-byte size limit; subsequent events are dropped", r.path, r.maxBytes)
-	}
 	n, err := r.file.Write(line)
 	if err == nil && n != len(line) {
 		err = io.ErrShortWrite
 	}
-	if err == nil {
-		r.size += int64(n)
+	if err != nil {
+		return err
 	}
-	return err
+	if r.seen == nil {
+		r.seen = make(map[string]struct{})
+	}
+	r.seen[key] = struct{}{}
+	return nil
 }
 
 func (r *JSONLRecorder) Close() error {

core/output/jsonl_test.go

@@ -4,12 +4,12 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
-	"strings"
 	"sync"
 	"testing"
 
 	aop "github.com/chainreactors/aiscan/aop"
 	"github.com/chainreactors/aiscan/core/eventbus"
+	"google.golang.org/protobuf/proto"
 )
 
 func TestJSONLRecorderWritesConcurrentEventsAsCompleteLines(t *testing.T) {
@@ -77,74 +77,76 @@ func TestJSONLRecorderSwitchesFilesWithoutReplayingHistory(t *testing.T) {
 	}
 }
 
-func TestJSONLRecorderStopsWritingAtSizeLimit(t *testing.T) {
-	path := filepath.Join(t.TempDir(), "capped.jsonl")
+func TestJSONLRecorderSkipsDuplicateEventIDWithinSession(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "events.jsonl")
 	bus := eventbus.New[*aop.Event]()
 	recorder, err := NewJSONLRecorder(bus, path)
 	if err != nil {
 		t.Fatal(err)
 	}
-	recorder.mu.Lock()
-	recorder.maxBytes = 512
-	recorder.mu.Unlock()
-
-	const emitted = 100
-	for i := 0; i < emitted; i++ {
-		bus.Emit(jsonlTestMessage(fmt.Sprintf("event-%d", i)))
+	event := &aop.Event{Id: "event-retry", SessionId: "session-1", Payload: &aop.Event_Status{Status: &aop.Status{State: "ready"}}}
+	bus.Emit(event)
+	bus.Emit(proto.Clone(event).(*aop.Event))
+	if err := recorder.Close(); err != nil {
+		t.Fatal(err)
 	}
+	events, err := ReadJSONL(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(events) != 1 {
+		t.Fatalf("JSONL events = %d, want 1", len(events))
+	}
+}
 
-	info, err := os.Stat(path)
+func TestJSONLRecorderLoadsExistingEventIDsAcrossRestart(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "events.jsonl")
+	event := jsonlTestMessage("persisted")
+	bus := eventbus.New[*aop.Event]()
+	first, err := NewJSONLRecorder(bus, path)
 	if err != nil {
 		t.Fatal(err)
 	}
-	// The cap bounds the payload; only the one marker line may exceed it.
-	if info.Size() > 512+256 {
-		t.Fatalf("file size = %d, want bounded near 512", info.Size())
+	bus.Emit(event)
+	if err := first.Close(); err != nil {
+		t.Fatal(err)
 	}
-	data, err := os.ReadFile(path)
+	second, err := NewJSONLRecorder(bus, path)
 	if err != nil {
 		t.Fatal(err)
 	}
-	if !strings.Contains(string(data), "size limit reached") {
-		t.Fatalf("truncation marker missing:\n%s", data)
+	bus.Emit(proto.Clone(event).(*aop.Event))
+	if err := second.Close(); err != nil {
+		t.Fatal(err)
 	}
-	// The marker is a non-JSON comment line: the file must stay readable.
 	events, err := ReadJSONL(path)
 	if err != nil {
 		t.Fatal(err)
 	}
-	if len(events) == 0 || len(events) >= emitted {
-		t.Fatalf("persisted events = %d, want partial prefix of %d", len(events), emitted)
-	}
-	if err := recorder.Close(); err == nil || !strings.Contains(err.Error(), "size limit") {
-		t.Fatalf("Close() error = %v, want size limit error", err)
+	if len(events) != 1 {
+		t.Fatalf("events after recorder restart = %d, want 1", len(events))
 	}
 }
 
-func TestJSONLRecorderSwitchResetsSizeBudget(t *testing.T) {
-	dir := t.TempDir()
-	bus := eventbus.New[*aop.Event]()
-	recorder, err := NewJSONLRecorder(bus, filepath.Join(dir, "first.jsonl"))
-	if err != nil {
+func TestScanJSONLRejectsNonEventLines(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "invalid.jsonl")
+	if err := os.WriteFile(path, []byte("traffic\n"), 0o644); err != nil {
 		t.Fatal(err)
 	}
-	recorder.mu.Lock()
-	recorder.maxBytes = 256
-	recorder.mu.Unlock()
-	for i := 0; i < 10; i++ {
-		bus.Emit(jsonlTestMessage(fmt.Sprintf("first-%d", i)))
-	}
-	second := filepath.Join(dir, "second.jsonl")
-	if err := recorder.Switch(second); err != nil {
-		t.Fatal(err)
+	if _, err := ReadJSONL(path); err == nil {
+		t.Fatal("ReadJSONL accepted a non-event line")
 	}
-	bus.Emit(jsonlTestMessage("after-switch"))
-	events, err := ReadJSONL(second)
+}
+
+func TestJSONLRecorderRejectsEventsWithoutIdentity(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "events.jsonl")
+	recorder, err := NewJSONLRecorder(eventbus.New[*aop.Event](), path)
 	if err != nil {
 		t.Fatal(err)
 	}
-	if len(events) != 1 || events[0].Id != "after-switch" {
-		t.Fatalf("second file events = %#v, want the post-switch event", events)
+	defer recorder.Close()
+	if err := recorder.Write(&aop.Event{SessionId: "session", Payload: &aop.Event_Status{Status: &aop.Status{State: "ready"}}}); err == nil {
+		t.Fatal("Write accepted an event without an id")
 	}
 }
 

core/output/render_test.go

@@ -127,7 +127,11 @@ func TestEventRendererSeparatesContinuationSessionHeaders(t *testing.T) {
 }
 
 func renderEvent(event *aop.Event) *aop.Event {
-	event.Id = "e-1"
+	if message := event.GetMessage(); message != nil && message.Id != "" {
+		event.Id = "e-" + message.Id
+	} else {
+		event.Id = "e-session-started"
+	}
 	event.SessionId = "session-1"
 	event.TurnId = "turn-1"
 	event.Emitter = "aiscan"

core/resources/resources.go

@@ -122,6 +122,14 @@ func Init(ctx context.Context, opts Options) (*Set, error) {
 	set.FingersConfig = fingers.NewConfig()
 	set.FingersConfig.FullFingers = finalFullFingers
 	set.NeutronConfig = neutron.NewConfig().WithTemplates(finalTemplates)
+	// Neutron compiles each template's HTTP transport when NewEngine runs.
+	// Carry the caller's egress proxy into the config before compilation so
+	// embedded and remote templates cannot bypass the Runner Hub. The engine
+	// package still applies the process default for compatibility with callers
+	// that construct neutron commands directly.
+	if opts.Proxy != "" {
+		set.NeutronConfig.WithProxy(opts.Proxy)
+	}
 
 	set.Fingers, err = fingers.NewEngineWithFingers(finalFullFingers)
 	if err != nil {