remove chatwait file from 9p filesystem
This commit is contained in:
parent
182682cbb7
commit
b4237c326c
|
|
@ -26,17 +26,13 @@ type Session struct {
|
||||||
cancel context.CancelFunc
|
cancel context.CancelFunc
|
||||||
log []byte
|
log []byte
|
||||||
logVers uint32
|
logVers uint32
|
||||||
logMu sync.Mutex
|
|
||||||
logCond *sync.Cond
|
|
||||||
ChatOffset int
|
ChatOffset int
|
||||||
plan []byte
|
plan []byte
|
||||||
prevPrompt []byte // last submitted prompt; overwritten on each new submission
|
prevPrompt []byte // last submitted prompt; overwritten on each new submission
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewSession(id string, core agent.Core, ctx context.Context, cancel context.CancelFunc) *Session {
|
func NewSession(id string, core agent.Core, ctx context.Context, cancel context.CancelFunc) *Session {
|
||||||
s := &Session{id: id, Core: core, Ctx: ctx, cancel: cancel}
|
return &Session{id: id, Core: core, Ctx: ctx, cancel: cancel}
|
||||||
s.logCond = sync.NewCond(&s.logMu)
|
|
||||||
return s
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (sess *Session) RunnableID() string { return sess.id }
|
func (sess *Session) RunnableID() string { return sess.id }
|
||||||
|
|
@ -58,11 +54,9 @@ func (sess *Session) AppendLog(data []byte) {
|
||||||
sess.log = append(sess.log, data...)
|
sess.log = append(sess.log, data...)
|
||||||
sess.logVers++
|
sess.logVers++
|
||||||
sess.mu.Unlock()
|
sess.mu.Unlock()
|
||||||
sess.logCond.Broadcast()
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// EnsureTrailingNewline appends a newline if the log doesn't already end with one.
|
// EnsureTrailingNewline appends a newline if the log doesn't already end with one.
|
||||||
// Always signals waiters to unblock chatwait at end of turn.
|
|
||||||
func (sess *Session) EnsureTrailingNewline() {
|
func (sess *Session) EnsureTrailingNewline() {
|
||||||
sess.mu.Lock()
|
sess.mu.Lock()
|
||||||
if len(sess.log) > 0 && sess.log[len(sess.log)-1] != '\n' {
|
if len(sess.log) > 0 && sess.log[len(sess.log)-1] != '\n' {
|
||||||
|
|
@ -70,47 +64,6 @@ func (sess *Session) EnsureTrailingNewline() {
|
||||||
}
|
}
|
||||||
sess.logVers++
|
sess.logVers++
|
||||||
sess.mu.Unlock()
|
sess.mu.Unlock()
|
||||||
sess.logCond.Broadcast()
|
|
||||||
}
|
|
||||||
|
|
||||||
// WaitLog blocks until the log grows past offset, then returns the new bytes.
|
|
||||||
// Returns nil if ctx is cancelled or the log is signalled without growth
|
|
||||||
// (e.g. end of turn).
|
|
||||||
func (sess *Session) WaitLog(ctx context.Context, offset int) []byte {
|
|
||||||
stop := context.AfterFunc(ctx, func() {
|
|
||||||
sess.logCond.Broadcast()
|
|
||||||
})
|
|
||||||
defer stop()
|
|
||||||
|
|
||||||
sess.logMu.Lock()
|
|
||||||
defer sess.logMu.Unlock()
|
|
||||||
|
|
||||||
// Snapshot version before waiting.
|
|
||||||
sess.mu.RLock()
|
|
||||||
startVers := sess.logVers
|
|
||||||
sess.mu.RUnlock()
|
|
||||||
|
|
||||||
for ctx.Err() == nil {
|
|
||||||
sess.mu.RLock()
|
|
||||||
n := len(sess.log)
|
|
||||||
v := sess.logVers
|
|
||||||
var data []byte
|
|
||||||
if n > offset {
|
|
||||||
data = make([]byte, n-offset)
|
|
||||||
copy(data, sess.log[offset:])
|
|
||||||
}
|
|
||||||
sess.mu.RUnlock()
|
|
||||||
if data != nil {
|
|
||||||
return data
|
|
||||||
}
|
|
||||||
// If version changed but no new content (e.g. end-of-turn signal),
|
|
||||||
// return nil to unblock the caller.
|
|
||||||
if v != startVers {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
sess.logCond.Wait()
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// LogInfo returns the current log length and version atomically.
|
// LogInfo returns the current log length and version atomically.
|
||||||
|
|
|
||||||
|
|
@ -36,7 +36,6 @@ var SessionFileList = []struct {
|
||||||
{"env", 0444},
|
{"env", 0444},
|
||||||
{"tail", 0555},
|
{"tail", 0555},
|
||||||
{"prompt.prev", 0444},
|
{"prompt.prev", 0444},
|
||||||
{"chatwait", 0444},
|
|
||||||
{"context", 0444},
|
{"context", 0444},
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -194,28 +193,6 @@ func (h *sessionHelper) fileSpec(name string, mode os.FileMode) FileSpec {
|
||||||
return []byte(v + "\n"), v, nil
|
return []byte(v + "\n"), v, nil
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if name == "chatwait" {
|
|
||||||
fs.Wait = func(connCtx context.Context, base string) ([]byte, string, error) {
|
|
||||||
ctx, cancel := context.WithCancel(connCtx)
|
|
||||||
defer cancel()
|
|
||||||
context.AfterFunc(h.sess.Ctx, cancel)
|
|
||||||
if h.sess.Core.State() == "idle" {
|
|
||||||
return nil, base, nil
|
|
||||||
}
|
|
||||||
offset := 0
|
|
||||||
if base == "" {
|
|
||||||
n, _ := h.sess.LogInfo()
|
|
||||||
offset = n
|
|
||||||
} else {
|
|
||||||
offset, _ = strconv.Atoi(base)
|
|
||||||
}
|
|
||||||
data := h.sess.WaitLog(ctx, offset)
|
|
||||||
if data == nil {
|
|
||||||
return nil, strconv.Itoa(offset), nil
|
|
||||||
}
|
|
||||||
return data, strconv.Itoa(offset + len(data)), nil
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return fs
|
return fs
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -448,60 +448,6 @@ func TestSessionAppendLog(t *testing.T) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestSessionWaitLog(t *testing.T) {
|
|
||||||
sess := testSession("wl")
|
|
||||||
defer sess.Cancel()
|
|
||||||
|
|
||||||
sess.AppendLog([]byte("existing"))
|
|
||||||
|
|
||||||
// WaitLog should return new content appended after offset.
|
|
||||||
done := make(chan []byte, 1)
|
|
||||||
go func() {
|
|
||||||
data := sess.WaitLog(context.Background(), 8) // offset = len("existing")
|
|
||||||
done <- data
|
|
||||||
}()
|
|
||||||
|
|
||||||
// Give goroutine time to block.
|
|
||||||
time.Sleep(10 * time.Millisecond)
|
|
||||||
sess.AppendLog([]byte(" new"))
|
|
||||||
|
|
||||||
select {
|
|
||||||
case data := <-done:
|
|
||||||
if string(data) != " new" {
|
|
||||||
t.Errorf("WaitLog = %q; want %q", data, " new")
|
|
||||||
}
|
|
||||||
case <-time.After(time.Second):
|
|
||||||
t.Fatal("WaitLog did not unblock")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestSessionWaitLogEndOfTurn(t *testing.T) {
|
|
||||||
sess := testSession("wl2")
|
|
||||||
defer sess.Cancel()
|
|
||||||
|
|
||||||
sess.AppendLog([]byte("content\n"))
|
|
||||||
|
|
||||||
// WaitLog should unblock on EnsureTrailingNewline (end-of-turn signal)
|
|
||||||
// even when no new content is added.
|
|
||||||
done := make(chan []byte, 1)
|
|
||||||
go func() {
|
|
||||||
data := sess.WaitLog(context.Background(), 8) // offset = len("content\n")
|
|
||||||
done <- data
|
|
||||||
}()
|
|
||||||
|
|
||||||
time.Sleep(10 * time.Millisecond)
|
|
||||||
sess.EnsureTrailingNewline() // no new byte, but signals end of turn
|
|
||||||
|
|
||||||
select {
|
|
||||||
case data := <-done:
|
|
||||||
if data != nil {
|
|
||||||
t.Errorf("WaitLog = %q; want nil", data)
|
|
||||||
}
|
|
||||||
case <-time.After(time.Second):
|
|
||||||
t.Fatal("WaitLog did not unblock on end-of-turn")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestSessionStoreFileMode(t *testing.T) {
|
func TestSessionStoreFileMode(t *testing.T) {
|
||||||
if m, ok := store.SessionStoreFileMode("new"); !ok || m != 0666 {
|
if m, ok := store.SessionStoreFileMode("new"); !ok || m != 0666 {
|
||||||
t.Errorf("SessionStoreFileMode(new) = %o, %v", m, ok)
|
t.Errorf("SessionStoreFileMode(new) = %o, %v", m, ok)
|
||||||
|
|
|
||||||
Reference in New Issue