diff --git a/docs/changes/unreleased/1514-teams-behaviour.md b/docs/changes/unreleased/1514-teams-behaviour.md new file mode 100644 index 000000000..d91db8457 --- /dev/null +++ b/docs/changes/unreleased/1514-teams-behaviour.md @@ -0,0 +1,14 @@ +--- +kind: fixed +title: one wake per member reply, team_stop reaches background members, a demoted manager loses its verbs +pr: 1514 +surface: [chat, engine, remote] +invalidates: + - "A member's reply to its manager woke the manager, and the finished event of the same member turn woke it again seconds later, so the ten-wake loop breaker tripped after about five real rounds. One reply is now one wake, and ten wakes are ten replies." + - "team_stop did not stop a member codeaf opened in the background with no window on it; the tool and the manual said so. The member's own session now performs its manager's stop, and the tool and the manual say it stops." + - "A conversation removed as a team's manager kept every manager verb on its belt, each refusing when called (team.go said this was deliberate). The verbs now leave at its next step, and a remembered call is told the role went." + - "Closing a sub-team whose manager alone was working closed it at once while the manager kept running. The card now opens with Wrap up first leading, and Close now stops that manager's turn but keeps its tab." + - "A team's daily spend that could not be read counted as $0, so the cap stopped holding. A capped team now starts nothing new until the spend can be read, and says why." + - "A team's packet file rotation kept only waiting packets, so a decided answer lived one more rotation and was gone. Answers not yet handed to their asker, and today's cap decisions, are now carried across rotations." + - "Teams.Name and Teams.Propose with a zero or negative budget gave the engine's model call no deadline. Zero now means the engine's 30-second ceiling, and a negative budget makes no call." +--- diff --git a/docs/design/conversations-and-teams/DESIGN.md b/docs/design/conversations-and-teams/DESIGN.md index f10358e91..5947b470e 100644 --- a/docs/design/conversations-and-teams/DESIGN.md +++ b/docs/design/conversations-and-teams/DESIGN.md @@ -274,6 +274,9 @@ tool call reads nothing until it returns, and a directive is advice a model may is the person's own Stop: it ends the current turn, deletes nothing, and leaves background tasks and jobs running. +(Changed 2026-09-25: a member codeaf opened without a window now reads its manager's +stop from Traffic during a running turn and ends that turn too.) + **What it may not do.** Approve members' permission prompts. Those are the person's safety gate, and a manager that could answer them would make every approval rule meaningless. If that is ever wanted, it is a separate, explicit per-team setting. @@ -524,6 +527,13 @@ the person to type again is not running a team, so the lines that ask for an ans something else starts, and the Traffic says `opened @handle; this team's auto-wake is off, so no turn was started.` +(Changed 2026-09-25: a running managed member stats its team's Traffic once per +tick and reads it when it moves, to honor a manager's stop.) + +(Changed 2026-09-25: a member's reply and the finished event from that same turn +start one manager wake, even when the event arrives after the settle. Ten wakes +therefore mean ten such reply rounds.) + **Mentioning a team or a chat from the composer.** `@` is still the one list (`internal/tui3`'s `files.go`, `mention.go`). Its first row is the words team, chat and file, each a press that types `@team:`, `@chat:` or `@file:` and keeps that @@ -570,6 +580,9 @@ the Traffic's asking row and a note, not a question on the manager's tab. Over ` are off and say so. An unreadable teams file on the engine is not moved aside from a window over `--host`; the window holds no teams until it can be read. +(Changed 2026-09-25: a stop also reaches a background member with no window; +starting a new member still needs the window's door.) + **Not in v1.** Collision flags when two members touch the same files, and dispatch of whole plans. (Nested managers, a sub-team's manager a member of the parent team with reports flowing up and directives down, were built later: 8.9.) @@ -990,6 +1003,10 @@ it), and `Close team…`. Closable with `esc`. the teams page's pane while that page stands, so the rail beside them still says which team they are about; they covered the rail before. Elsewhere they are centred on the frame. +(Changed 2026-09-25: a manager of a team being closed is counted as working and +its current turn is stopped by Close now even if it is also in an open parent; +its tab stays open. Other shared conversations keep the prior rule.) + ### 8.6 Where this departs from the brief, and why - **The home flag resolves to the nearest manager up its chain**, so giving an unmanaged @@ -1093,6 +1110,10 @@ different crossing and a new packet. A decided `raise` lifts the ceiling to the person's own words hold until the day turns or the cap is changed. Spend is read through `TeamSpend` only when `TeamSpendStamp` moved (per pool, per session), never per model request. +(Changed 2026-09-25: if a capped pool's spend read fails, the start is held +with the read error, no cap packet is raised, and the failed reading is retried +on the next check. An uncapped pool does not read spend.) + **Wrap up first: the door and the marker (for d2).** The interface appends ONE Traffic entry to the team's log: @@ -1129,6 +1150,11 @@ packet whole, trail and escalated state included). The reader folds the rotated current one; a carry replaces what the older file said of that id. A waiting packet is never lost; a decided one stays readable for one more rotation. +(Changed 2026-09-25: a decided answer not yet handed to its raiser is carried +across rotations, newest first within half the rotation size. Today's cap packets +are carried too. Delivery appends a `told` line; after that the answer has the +former one-rotation lifetime. Conflicts have their own Traffic directives.) + **Over `--host`.** All of the above runs where the conversations run, the engine: questions, caps, the wrap-up clock and the verbs are the engine's session reading the engine's profile and ledger, so a window over `--host` needs nothing new. The clock travels with the team: diff --git a/internal/manual/chat/team-manager.md b/internal/manual/chat/team-manager.md index a45fb814b..ae3ff5782 100644 --- a/internal/manual/chat/team-manager.md +++ b/internal/manual/chat/team-manager.md @@ -159,8 +159,8 @@ message. With auto-wake off the conversation is still opened behind the one you no turn is started; the traffic says `opened @lexer; this team's auto-wake is off, so no turn was started. It reads the brief when it next runs.`, and the brief arrives on that next turn the same way. When the manager stops a member, -this window stops it the way your own Stop would. Both happen only in a window that has those -conversations open. +its current turn ends the way your own Stop would, including when codeaf opened it in the +background without a window. Starting a new member still needs a window holding the manager. ## Who a Traffic row is from and who it is to @@ -221,7 +221,7 @@ on one waits for you. | `team_status` | every member's handle, title and state (running, asking, idle, failed), the question waiting, the files touched, and recent traffic | no | | `team_read` | the end of one member's conversation, bounded; the member is not told | no | | `team_send` | a message to one member, to several (one message, every handle in `to`), or to everyone, as a note (information, which waits) or a directive (an instruction, which starts an idle member) | no | -| `team_stop` | ends one member's current turn, the way your own Stop does: nothing is deleted, and its background tasks and jobs keep running. It is carried out by a window that has the member open; a member codeaf opened in the background, with no window on it, is not stopped and its turn runs to its end | no | +| `team_stop` | ends one member's current turn, the way your own Stop does, whether a window has it open or codeaf opened it in the background: nothing is deleted, and its background tasks and jobs keep running | no | | `team_start` | a new member conversation with a handle and a brief; it opens in the team's folder and is handed the brief, marked as the manager's, on its first request. With kind `team` it starts a sub-team instead (see **Sub-teams**) | yes | | `team_decide` | answers a decision packet waiting on the manager, most often a member's question: an option, or its own words | no | | `team_escalate` | sends a packet waiting on the manager up, to its own manager or to you, with the reason it is not the manager's to decide | no | @@ -234,6 +234,17 @@ every one of them is logged in the team's traffic. Questions, packets, caps and on the page **Team questions, decisions and caps**. Like any tool, each can be set to ask or allow in `/settings` under the tool approvals. +## What happens when a conversation stops being the manager + +At its next step, a conversation removed as manager loses `team_status`, `team_read`, +`team_send`, `team_stop`, `team_start`, `team_decide`, `team_escalate` and +`team_close_report`. If it is still a member of a managed team, it keeps `team_post` and +`team_raise`; outside a team it loses those too. A remembered call to a removed manager +tool says `team_send is no longer one of your tools: this conversation no longer manages a +team.` A removed member tool says `team_post is no longer one of your tools: this +conversation is no longer a member of a team with a manager.` Made a manager again, it gets +the manager tools back at its next step. + ## The member's verbs A member of a team that has a manager has `team_post`: a message to the room (every member and @@ -348,9 +359,10 @@ working, that is straight away. When it is idle, it depends on the kind of messa - A **directive** starts an idle member's turn. The member is handed the directive, marked `◆ directive from manager`, never as if you had typed it. - A **note** wakes nobody. An idle member reads it when it next runs, for whatever reason. -- A member's **reply to the manager** (`team_post` to the manager), and a member finishing, - failing or starting to wait on you, start an idle manager's turn. Replies that arrive within - a few seconds of each other are gathered into one turn rather than one turn each. +- A member's **reply to the manager** (`team_post` to the manager), and a member finishing + without a reply, failing or starting to wait on you, start an idle manager's turn. A reply + and the end of the same turn are one wake, even if the ending arrives after the manager + already ran. Replies arriving within a few seconds are gathered into one turn. A member no window has open is opened by codeaf in the background so it can run, and a window that opens it later joins the running conversation. When that cannot be done, the traffic says @@ -361,6 +373,7 @@ conversation is running. A wake spends through the same limits a turn you start more bound it: one conversation is woken at most 20 times an hour, and a manager woken 10 times by its team with nothing from you stops being woken and asks you instead, as a waiting line in the traffic. It is woken again after you next say something to it. +The ten wakes count ten replies when each member turn replies and then finishes. A team's auto-wake can be turned off. `team messages wake` in `/settings` under **Teams** is the default every team inherits (on), and a team can override it for itself and the teams diff --git a/internal/manual/chat/team-questions-and-caps.md b/internal/manual/chat/team-questions-and-caps.md index fdc2607ce..648882df6 100644 --- a/internal/manual/chat/team-questions-and-caps.md +++ b/internal/manual/chat/team-questions-and-caps.md @@ -43,8 +43,15 @@ when it has no manager above it. Your answer is handed to the manager marked ` and wakes it. Packets are kept in the profile of the machine the conversations run on, in -`teams//decisions.jsonl` for the team each was raised from. Past a megabyte the file starts -a new one, keeping every packet still waiting. +`teams//decisions.jsonl` for the team each was raised from. + +## Do answered team questions survive packet file rotation + +Past a megabyte the packet file starts a new one. It keeps every packet still waiting, +today's cap decisions, and decided answers not yet handed to their raisers, newest first +within half the rotation size. Older owed answers beyond that bound remain readable for one +more rotation. Once a raiser has been handed an answer, codeaf records that and keeps it +readable for one more rotation. ## A team's daily cap @@ -66,6 +73,14 @@ When the pool reaches its cap: A manager can never raise a cap: money is yours. Every held wake is one line in the traffic, `held @web: harbor reached its $5 cap today`. +## What if today's team spending cannot be read + +When a capped team's usage ledger or team list cannot be read, no new paid team work starts. +The Traffic says `held @web: harbor has a $5 daily cap and today's spend could not be read +(), so nothing new starts until it can be read`. `team_start` gives the same reason +after `No new member starts:`. No cap card is raised against an unreadable figure. The next +check reads again, and an uncapped team does not read the ledger for this check. + ## Two windows ask once when a team reaches its cap You are asked once for that team, that day, and that ceiling. A second codeaf window, or a diff --git a/internal/manual/chat/teams-page.md b/internal/manual/chat/teams-page.md index 279960b45..17772ddb5 100644 --- a/internal/manual/chat/teams-page.md +++ b/internal/manual/chat/teams-page.md @@ -302,11 +302,13 @@ away; `←` `→` choose a colour. `esc` or `Done` puts the card away. **Close now** and **Cancel**. When the team has a manager, `Wrap up first` leads: the manager is asked to have everyone finish and commit and to bring you a closing report, which arrives as a card on this page with `Close` and `Keep going`; the team closes when you choose Close. - `Close now` stops every member's turn and closes their tabs at once. `Cancel` or `esc` + A working manager is named on the card even when it is also in a team above this one. + `Close now` stops every member's turn, including that manager's, and closes tabs belonging + only to the closing teams. A manager also in an open team above keeps its tab. `Cancel` or `esc` changes nothing. -Closing a team closes the teams under it. A conversation that is also in another open team is -never stopped by the close. The conversation you are looking at keeps its tab, so a close never +Closing a team closes the teams under it. Other conversations that are also in another open team +are not stopped by the close. The conversation you are looking at keeps its tab, so a close never moves you. A closed team spends nothing, is not on the conversations view or the strip, and waits under `▸ Closed · N`. diff --git a/internal/manual/chat_test.go b/internal/manual/chat_test.go index 5f4cc9971..787031c2c 100644 --- a/internal/manual/chat_test.go +++ b/internal/manual/chat_test.go @@ -46,6 +46,7 @@ func TestTheChatManualAnswersTheQuestionsPeopleAsk(t *testing.T) { {"how do I see all my teams and what waits on me", "teams-page"}, {"does renaming a team update the message box", "teams-page"}, {"how do I close a team", "teams-page"}, + {"will closing a sub-team stop its manager if that manager is also in the parent team", "teams-page"}, {"the pane said the manager was open in another window", "teams-page"}, {"how do I reopen a closed team", "teams-page"}, {"where do I change one team's settings", "teams-page"}, @@ -68,10 +69,15 @@ func TestTheChatManualAnswersTheQuestionsPeopleAsk(t *testing.T) { {"can the manager answer a member's permission prompt", "team-manager"}, {"how does a member post to the room", "team-manager"}, {"does a directive wake an idle member", "team-manager"}, + {"why did my manager wake twice for one reply", "team-manager"}, + {"can a manager stop a member working in the background without a window", "team-manager"}, + {"what tools does a conversation lose when it stops being the manager", "team-manager"}, // Questions, packets, caps and wrapping up (team-questions-and-caps.md). {"does a member's question go to the manager or to me", "team-questions-and-caps"}, {"what is a decision packet", "team-questions-and-caps"}, {"what happens when a team reaches its daily cap", "team-questions-and-caps"}, + {"what if codeaf cannot read today's team spending", "team-questions-and-caps"}, + {"do answered team questions survive packet file rotation", "team-questions-and-caps"}, {"why did two windows both ask me about the team cap", "team-questions-and-caps"}, {"how do I wrap up a team before closing it", "team-questions-and-caps"}, {"what happens to a wrap-up when codeaf restarts", "team-questions-and-caps"}, diff --git a/internal/remote/teamask.go b/internal/remote/teamask.go index cbb6ee60b..b3de818ac 100644 --- a/internal/remote/teamask.go +++ b/internal/remote/teamask.go @@ -34,13 +34,16 @@ func teamAskKnown(agent any) bool { _, ok := agent.(teamAskDoor); return ok } // suggestion. const teamAskOffWord = "this engine cannot suggest team names or teams" -// teamAskWithin is the engine's bound on one ask: the wall's budget, or none -// when the wall sent none. +// teamAskCeiling bounds a peer's request while leaving the wall's five-second +// name wait and ten-second Organize wait in control of their own calls. +const teamAskCeiling = 30 * time.Second + +// teamAskWithin uses the shorter of the wall's budget and the engine's ceiling. func teamAskWithin(budget time.Duration) (context.Context, context.CancelFunc) { - if budget > 0 { - return context.WithTimeout(context.Background(), budget) + if budget == 0 || budget > teamAskCeiling { + budget = teamAskCeiling } - return context.WithCancel(context.Background()) + return context.WithTimeout(context.Background(), budget) } // teamAskCall answers the two asks from agent, and says whether the method was @@ -61,6 +64,9 @@ func teamAskCall(agent WrappedAgent, call Frame) (json.RawMessage, bool, error) if err != nil { return nil, true, err } + if args.Budget < 0 { + return nil, true, context.DeadlineExceeded + } ctx, cancel := teamAskWithin(args.Budget) defer cancel() name, err := door.NameTeam(ctx, args.Titles) @@ -74,6 +80,9 @@ func teamAskCall(agent WrappedAgent, call Frame) (json.RawMessage, bool, error) if err != nil { return nil, true, err } + if args.Budget < 0 { + return nil, true, context.DeadlineExceeded + } ctx, cancel := teamAskWithin(args.Budget) defer cancel() proposal, err := door.ProposeTeams(ctx, args.In) diff --git a/internal/remote/teamask_test.go b/internal/remote/teamask_test.go index c3520d5b4..cb6b9f39c 100644 --- a/internal/remote/teamask_test.go +++ b/internal/remote/teamask_test.go @@ -17,6 +17,7 @@ type askAgent struct { in session.TeamProposalInput deadline bool left time.Duration + calls int } func (a *askAgent) NameTeam(ctx context.Context, titles []string) (string, error) { @@ -37,6 +38,7 @@ func (a *askAgent) ProposeTeams(ctx context.Context, in session.TeamProposalInpu } func (a *askAgent) note(ctx context.Context) { + a.calls++ deadline, ok := ctx.Deadline() a.deadline = ok if ok { @@ -44,6 +46,56 @@ func (a *askAgent) note(ctx context.Context) { } } +func TestTeamAskBudgetBoundsEveryModelCall(t *testing.T) { + if teamAskCeiling <= 10*time.Second { + t.Fatalf("the engine ceiling %v cuts off the wall's Organize wait", teamAskCeiling) + } + for _, tc := range []struct { + name string + budget time.Duration + want time.Duration + }{ + {"zero", 0, teamAskCeiling}, + {"huge", time.Hour, teamAskCeiling}, + {"small", 250 * time.Millisecond, 250 * time.Millisecond}, + } { + for _, method := range []string{MethodTeamsName, MethodTeamsPropose} { + t.Run(tc.name+"/"+method, func(t *testing.T) { + far := &askAgent{fakeAgent: &fakeAgent{model: "m"}} + loop := askLoop(t, far) + var err error + if method == MethodTeamsName { + _, err = loop.Client.call(nil, method, TeamNameArgs{Budget: tc.budget}) + } else { + _, err = loop.Client.call(nil, method, TeamProposeArgs{Budget: tc.budget}) + } + if err != nil { + t.Fatal(err) + } + if far.calls != 1 || !far.deadline || far.left <= 0 || far.left > tc.want || far.left < tc.want-time.Second { + t.Fatalf("budget %v called %d times with deadline=%v, remaining=%v; want at most %v", tc.budget, far.calls, far.deadline, far.left, tc.want) + } + }) + } + } +} + +func TestExpiredTeamAskNeverCallsTheModel(t *testing.T) { + far := &askAgent{fakeAgent: &fakeAgent{model: "m"}} + loop := askLoop(t, far) + for _, method := range []string{MethodTeamsName, MethodTeamsPropose} { + var err error + if method == MethodTeamsName { + _, err = loop.Client.call(nil, method, TeamNameArgs{Budget: -time.Nanosecond}) + } else { + _, err = loop.Client.call(nil, method, TeamProposeArgs{Budget: -time.Nanosecond}) + } + if err == nil || err.Error() != context.DeadlineExceeded.Error() || far.calls != 0 { + t.Fatalf("expired %s called the model %d times and returned %v", method, far.calls, err) + } + } +} + func askLoop(t *testing.T, agent WrappedAgent) *Loop { t.Helper() loop, err := Loopback(Hello{Version: Version}, Options{Boot: func(Hello) (*Engine, error) { diff --git a/internal/remote/wire_teams.go b/internal/remote/wire_teams.go index b10d4ba84..cc0df08e3 100644 --- a/internal/remote/wire_teams.go +++ b/internal/remote/wire_teams.go @@ -116,14 +116,16 @@ const ( ) // TeamNameArgs is the titles a group is named from, and how long the wall -// will wait. Zero Budget is a caller with no deadline. +// will wait. Zero Budget uses the engine's ceiling; a negative Budget has +// already expired and never starts a model call. type TeamNameArgs struct { Titles []string `json:"titles"` Budget time.Duration `json:"budget,omitempty"` } // TeamProposeArgs is everything one Organize ask is about, and how long the -// wall will wait. +// wall will wait. Zero uses the engine's ceiling, and a negative Budget has +// already expired and never starts a model call. type TeamProposeArgs struct { In session.TeamProposalInput `json:"in"` Budget time.Duration `json:"budget,omitempty"` diff --git a/internal/session/agent.go b/internal/session/agent.go index 5e5f40e4f..e260c7c0c 100644 --- a/internal/session/agent.go +++ b/internal/session/agent.go @@ -1707,6 +1707,8 @@ func (a *Agent) startTurnLocked(ctx context.Context, user userMessage, watcher * a.notePersonTurn(user) a.rebindClientLocked(a.model) a.running = true + a.teamTurnAt = time.Now() + a.teamTurnSerial++ a.lastTurnTruncated = false // AND ANOTHER WINDOW HEARS ABOUT IT NOW rather than at the next heartbeat // (taskpresence.go). The nudge never blocks and never takes a lock, which is diff --git a/internal/session/connect.go b/internal/session/connect.go index 0c4f61c00..c8346b920 100644 --- a/internal/session/connect.go +++ b/internal/session/connect.go @@ -529,10 +529,11 @@ func (a *Agent) beltDefinitions() []ai.ToolDefinition { // this lesson twice). Appending costs exactly one invalidation, at the back, // once per family, and that is the price this design accepted in advance. // -// It is also why nothing here retires: the two connect tools stay on the belt +// Connected account tools do not retire here: the two connect tools stay on the belt // for the life of the session, and a model that asks for a family it already // holds is answered in one cheap line at the END of the transcript, which is -// free. Removing them would rewrite the block instead. +// free. Team verbs are the exception: a demotion removes a capability that +// cannot work, and that rare role change pays for the rewritten block. // // An empty answer means everything asked for was already there. func (a *Agent) armFamily(tools []bare.Tool) ([]string, error) { diff --git a/internal/session/loop.go b/internal/session/loop.go index 28817352c..d28741694 100644 --- a/internal/session/loop.go +++ b/internal/session/loop.go @@ -3618,6 +3618,9 @@ func (a *Agent) dispatchTool(ctx context.Context, ep *episode, hub *eventHub, ca if notice, withdrawn := a.withdrawalNotice(call.Function.Name); withdrawn { return toolResult{text: notice, isError: true, harness: true} } + if notice, retired := a.teamRetiredNotice(call.Function.Name); retired { + return toolResult{text: notice, isError: true, harness: true} + } // A name nobody ever had keeps the old answer, and keeps it word for word: // that one IS a sentence about the model. return toolResult{text: "Unknown tool: " + call.Function.Name, isError: true} diff --git a/internal/session/session.go b/internal/session/session.go index ab51dfc44..cf05c0ab8 100644 --- a/internal/session/session.go +++ b/internal/session/session.go @@ -2347,14 +2347,17 @@ type Agent struct { systemOwn bool // tools is the belt and definitions is its wire form, built once at // construction — rebuilding them per step would re-marshal every schema on - // the hot path — and thereafter APPEND-ONLY, under armMu (connect.go). + // the hot path — and thereafter grown only at the tail under armMu + // (connect.go), with one exception: the team verbs leave when the role that + // gave them does (team.go's [Agent.retireTeamTools]). // // Both are COPY-ON-WRITE: arming allocates a new array and swaps the header, // so a reader that took a snapshot under armMu may walk it without the lock // and can never see a half-written slice. Nothing already in either is ever - // moved, rewritten or removed, because the definition block rides at the - // front of every request and a definition that shifts re-bills the whole - // prompt behind it (internal/exec's tools.go states the law). + // moved or rewritten, and nothing but a team verb is removed, because the + // definition block rides at the front of every request and a definition that + // shifts re-bills the whole prompt behind it (internal/exec's tools.go states + // the law). tools []bare.Tool definitions []ai.ToolDefinition // served is what the belt cannot say about the tools an ACCOUNT named @@ -2397,6 +2400,9 @@ type Agent struct { // dispatcher that found a name missing needs to know whether it was taken or // never existed, and the two answers must not be able to disagree. withdrawn *toolWithdrawal + // teamRetired names team verbs a role change took away, so a remembered + // call gets the role's reason instead of being called an unknown tool. + teamRetired map[string]string // armMu guards those headers, those maps, the shelf and nothing else. It is not mu: // arming happens inside a tool call, and a tool call must never take the // lock Interrupt has to be able to take. @@ -2739,6 +2745,10 @@ type Agent struct { // the transcript, so without this bit a digest can only repeat the cut-off // prose and falsely make the node look complete. lastTurnTruncated bool + // teamTurnAt and teamTurnSerial identify the current turn for a manager's + // stop, so a late watch read cannot stop a later turn. + teamTurnAt time.Time + teamTurnSerial uint64 // memoryText is the block message[0] currently carries: what the // router asked for at the start of this turn, or the block a task node was // opened with (memory.go). It is under mu because it is rendered into the diff --git a/internal/session/stopcause.go b/internal/session/stopcause.go index 141d2f9c6..a9e1f365b 100644 --- a/internal/session/stopcause.go +++ b/internal/session/stopcause.go @@ -41,6 +41,10 @@ type StopDoor string const ( // StopByPerson is the stop key and nothing else: the person asked. StopByPerson StopDoor = "person stopped" + // StopByManager is a team manager's team_stop reaching this member's own + // session (team_wakewatch.go): the person's Stop in every respect but who + // asked, so the member's conversation says the manager did it. + StopByManager StopDoor = "stopped by its manager" // StopByTakeover is another window taking this conversation over // (takeover.go). The turn dies where it stands, mid-reply or not. StopByTakeover StopDoor = "taken over" @@ -155,6 +159,8 @@ func stopSentence(door StopDoor, name string) string { switch door { case StopByPerson: return "" + case StopByManager: + return "the manager stopped this turn" case StopByTakeover: return "this conversation was opened in another window, so the reply stopped here — ask again to pick it up" case StopByLeaving: diff --git a/internal/session/stopcause_test.go b/internal/session/stopcause_test.go index 4baa4bd9d..4f2152d21 100644 --- a/internal/session/stopcause_test.go +++ b/internal/session/stopcause_test.go @@ -121,7 +121,7 @@ func TestATurnThePersonStoppedSaysNothingAboutIt(t *testing.T) { // at nothing, and there is exactly one that is allowed to. func TestEveryDoorNamesItselfAndOnlyThePersonsIsSilent(t *testing.T) { doors := []StopDoor{ - StopByPerson, StopByTakeover, StopByLeaving, StopByClosing, + StopByPerson, StopByManager, StopByTakeover, StopByLeaving, StopByClosing, StopByAbandoned, StopByWorkStopped, StopByRetired, StopByEngineStopped, } seen := map[string]StopDoor{} diff --git a/internal/session/team.go b/internal/session/team.go index da539298a..69931b664 100644 --- a/internal/session/team.go +++ b/internal/session/team.go @@ -74,6 +74,7 @@ import ( "github.com/Agent-Field/agentfield/sdk/go/ai" "github.com/Agent-Field/codeaf/internal/config" "github.com/Agent-Field/codeaf/internal/exec/bare" + "github.com/Agent-Field/codeaf/internal/guard" "github.com/Agent-Field/codeaf/internal/teams" ) @@ -193,6 +194,9 @@ type teamSeat struct { // found by the delivery that handed it the brief and made good after it // (team_nest.go's [Agent.claimSubTeams]). claims []subTeamClaim + // told is the packet answers just handed over. Their durable marks are + // written after the seat lock is released and away from the turn's path. + told []string // handleTried says this process has asked for its handle once // (handlepick.go), so it is never asked for twice. handleTried bool @@ -425,14 +429,16 @@ func memberOf(team teams.Team, keys []string) (teams.Member, bool) { // handed back as one note, "" for nothing. It is called by [Agent.drainSteering] // before that drain takes the agent's lock. // -// THE VERBS ARRIVE BY THE ARMING DOOR AND NEVER LEAVE. A conversation made a -// manager halfway through its life is given the manager's verbs at its next -// boundary through [Agent.armFamily], the door a connected account and a loaded -// group arrive through, on the same append law: at the tail, nothing already -// there moves. A conversation that stops being a manager keeps the verbs and -// each one refuses, because it asks the file again when it is called -// ([Agent.teamTarget]); taking a tool out of the block would re-price the whole -// conversation for a verb nobody will call. +// THE VERBS ARRIVE BY THE ARMING DOOR AND LEAVE WITH THE ROLE. A conversation +// made a manager halfway through its life is given the manager's verbs at its +// next boundary through [Agent.armFamily], the door a connected account and a +// loaded group arrive through, on the same append law: at the tail, nothing +// already there moves. A conversation that stops being a manager loses them at +// its next boundary ([Agent.retireTeamTools]). They once stayed, each refusing +// when called, to spare the one re-pricing of the prompt a rewritten tool block +// costs; but a verb that cannot work must be absent, not broken (CLAUDE.md), +// and a demotion is rare enough that paying once for it is cheap. A role +// gained again appends its verbs at the tail as before. func (a *Agent) teamBoundary() string { profile := a.config.teamProfile() if profile == "" { @@ -453,7 +459,16 @@ func (a *Agent) teamBoundary() string { roles = a.teamRolesLocked(profile) } role := teamRoleBlock(roles, a.team.file) + told := append([]string(nil), a.team.told...) + a.team.told = nil a.team.mu.Unlock() + if len(told) > 0 { + guard.Go("team answers handed over", func() { + for _, id := range told { + _ = teams.Told(profile, id) + } + }) + } // A wrap-up the delivery just started is written down here, after the // seat's lock, so a restart keeps the clock (team_wrapup.go). a.teamWrapUpNote(profile) @@ -484,12 +499,97 @@ func (a *Agent) armTeamTools(roles []teamRole) { if manager || member { arriving = append(arriving, a.raiseTools()...) } + allowed := make(map[string]bool, len(arriving)) + for _, tool := range arriving { + allowed[tool.Name] = true + } + a.retireTeamTools(allowed) if len(arriving) == 0 { return } // A schema that will not parse is a bug in this build, and every tool here // is a literal; the error has nowhere useful to go at a boundary. _, _ = a.armFamily(arriving) + a.armMu.Lock() + for name := range allowed { + delete(a.teamRetired, name) + } + a.armMu.Unlock() +} + +// retireTeamTools takes off the belt every team verb the current roles do not +// give, and remembers why, so a model that calls one from memory is told the +// role changed rather than that the verb never existed ([Agent.teamRetiredNotice]). +// +// A BOUNDARY THAT CHANGES NOTHING ALLOCATES NOTHING: it runs at every step of +// every conversation in a profile, and nearly every time the belt already +// matches the role. When a verb does leave, the belt is rebuilt into fresh +// arrays under armMu, as [Agent.armFamily] grows it, so a reader holding the +// previous snapshot is never written under, and every tool that stays keeps +// its order. +func (a *Agent) retireTeamTools(allowed map[string]bool) { + a.armMu.Lock() + defer a.armMu.Unlock() + leaving := false + for _, tool := range a.tools { + if teamToolName(tool.Name) && !allowed[tool.Name] { + leaving = true + break + } + } + if !leaving { + return + } + kept := make([]bare.Tool, 0, len(a.tools)) + for _, tool := range a.tools { + if !teamToolName(tool.Name) || allowed[tool.Name] { + kept = append(kept, tool) + continue + } + if a.teamRetired == nil { + a.teamRetired = map[string]string{} + } + a.teamRetired[tool.Name] = teamRetiredMember + if teamManagerToolName(tool.Name) { + a.teamRetired[tool.Name] = teamRetiredManager + } + } + a.tools = kept + a.definitions, _ = toolDefinitions(kept) +} + +// The reasons a retired team verb is answered with, one per kind of verb. +const ( + teamRetiredManager = "this conversation no longer manages a team" + teamRetiredMember = "this conversation is no longer a member of a team with a manager" +) + +// teamToolName reports whether name is one of the team verbs. +func teamToolName(name string) bool { + for _, candidate := range teamToolNames { + if name == candidate { + return true + } + } + return false +} + +// teamManagerToolName reports whether name is a verb only a manager is given: +// every team verb but a member's post and its conflict raise. +func teamManagerToolName(name string) bool { + return teamToolName(name) && name != teamPostToolName && name != teamRaiseToolName +} + +// teamRetiredNotice is what a call to a team verb the role took away is +// answered with, and false for every other name. +func (a *Agent) teamRetiredNotice(name string) (string, bool) { + a.armMu.Lock() + reason, ok := a.teamRetired[name] + a.armMu.Unlock() + if !ok { + return "", false + } + return name + " is no longer one of your tools: " + reason + ".", true } // teamNewsLocked reads each team's Traffic past this conversation's cursor and @@ -614,7 +714,11 @@ func (a *Agent) firstTeamCursor(profile string, role teamRole) string { func (a *Agent) teamEntryLineLocked(profile string, role teamRole, entry teams.Entry) string { switch { case entry.Kind == teams.KindPacket: - return teamPacketLine(profile, role, entry) + line, told := teamPacketLine(profile, role, entry) + if told != "" { + a.team.told = append(a.team.told, told) + } + return line case entry.Kind == teams.KindStart && entry.Team != "": line := teamBriefLine(role, entry) if line == "" { @@ -625,6 +729,14 @@ func (a *Agent) teamEntryLineLocked(profile string, role teamRole, entry teams.E case role.manager && teams.IsWrapUp(entry): a.teamWrapUpBeginLocked(role, entry.At) return wrapUpLine(role, entry) + case role.manager && entry.Kind == teams.KindEvent && entry.State == teams.StateFinished && + teamFinishedAfterReply(profile, role.id, entry): + // A reply's ending starts no second wake, but the manager still reads + // that ending at its next boundary after the earlier wake has run. + lines := teamEventLines([]teams.Entry{entry}) + if len(lines) > 0 { + return lines[0] + } } return teamLine(role, entry) } diff --git a/internal/session/team_cap.go b/internal/session/team_cap.go index efd144211..5ff4f033d 100644 --- a/internal/session/team_cap.go +++ b/internal/session/team_cap.go @@ -64,17 +64,17 @@ type spendMemo struct { // teamPoolSpend is the pool owner's spend today, read again only when its // stamp moved. -func (a *Agent) teamPoolSpend(profile, owner, day string) float64 { +func (a *Agent) teamPoolSpend(profile, owner, day string) (float64, error) { stamp := teamSpendStamp(profile, owner, day) a.team.mu.Lock() if held, ok := a.team.spends[owner]; ok && held.stamp == stamp { a.team.mu.Unlock() - return held.usd + return held.usd, nil } a.team.mu.Unlock() spend, err := teamSpendOf(profile, owner, day) if err != nil { - return 0 + return 0, err } a.team.mu.Lock() if a.team.spends == nil { @@ -82,7 +82,7 @@ func (a *Agent) teamPoolSpend(profile, owner, day string) float64 { } a.team.spends[owner] = spendMemo{stamp: stamp, usd: spend.USD} a.team.mu.Unlock() - return spend.USD + return spend.USD, nil } // capPool is the pool team id draws on: its owner and the cap, 0 for none. @@ -141,7 +141,10 @@ func (a *Agent) teamCapHold(profile string, roles []teamRole) string { // poolHold is why owner's pool is held, "" when it is not. func (a *Agent) poolHold(profile string, owner teams.Team, cap float64) string { day := teamToday() - spent := a.teamPoolSpend(profile, owner.ID, day) + spent, err := a.teamPoolSpend(profile, owner.ID, day) + if err != nil { + return fmt.Sprintf("%s has a %s daily cap and today's spend could not be read (%s), so nothing new starts until it can be read", owner.Name, teamMoney(cap), oneLineTeam(err.Error())) + } latest, found := latestCapPacket(profile, owner.ID, day) ceiling := cap if found && latest.State == teams.PacketDecided && latest.Decision == teams.OptionRaiseCap && latest.Cap.RaiseTo > ceiling { diff --git a/internal/session/team_delegation_test.go b/internal/session/team_delegation_test.go index 999e15d43..0089426a2 100644 --- a/internal/session/team_delegation_test.go +++ b/internal/session/team_delegation_test.go @@ -9,9 +9,11 @@ package session import ( "context" "encoding/json" + "errors" "os" "path/filepath" "strings" + "sync/atomic" "testing" "time" @@ -91,6 +93,45 @@ func TestTeamQuestionGoesUpAsAPacketAndTheAnswerComesBack(t *testing.T) { } } +func TestTeamAnswerDeliveredAfterPacketRotationsIsMarkedTold(t *testing.T) { + fixture := newTeamFixture(t, true) + web := teamAgent(t, fixture, fixture.web, nil, nil) + web.teamBoundary() + p, err := teams.Raise(fixture.profile, teams.Packet{Team: fixture.teamID, Kind: teams.PacketQuestion, RaisedBy: "web", Question: "JSON or form data?"}) + if err != nil { + t.Fatal(err) + } + if _, err := teams.Decide(fixture.profile, p.ID, teams.FromManager, "JSON", "the other endpoints use it"); err != nil { + t.Fatal(err) + } + for i := 0; i < 6; i++ { + other, err := teams.Raise(fixture.profile, teams.Packet{Team: fixture.teamID, Kind: teams.PacketConflict, RaisedBy: "parser", + Question: strings.Repeat("unrelated work ", 30000), Options: []teams.Option{{Label: "go", Consequence: "carry on"}}}) + if err != nil { + t.Fatal(err) + } + if _, err := teams.Decide(fixture.profile, other.ID, teams.FromManager, "1", "done"); err != nil { + t.Fatal(err) + } + } + if got := web.teamBoundary(); !strings.Contains(got, "◆ answered: JSON") { + t.Fatalf("a rotated packet did not hand over its answer: %q", got) + } + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + got, err := teams.PacketByID(fixture.profile, p.ID) + if err == nil && got.Told { + raw, err := os.ReadFile(teams.DecisionsPath(fixture.profile, fixture.teamID)) + if err != nil || !strings.Contains(string(raw), `"op":"told"`) || !strings.Contains(string(raw), `"id":"`+p.ID+`"`) { + t.Fatalf("the packet was marked told without its append-only line: %v, %q", err, raw) + } + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatal("the delivery did not append a told line") +} + func last(log []teams.Entry) teams.Entry { return log[len(log)-1] } // PERMISSION PROMPTS AND QUESTIONS WITH questions_up OFF ARE THE PERSON'S. @@ -320,6 +361,91 @@ func TestTeamNoCapReadsNoSpend(t *testing.T) { } } +func TestTeamUnreadableSpendHoldsCappedWorkWithoutAPacket(t *testing.T) { + fastTeamWake(t) + fixture := newWakingTeamFixture(t) + five := 5.0 + if err := teams.Update(fixture.profile, func(f *teams.File) error { + return f.SetSettings(fixture.teamID, func(s *teams.Settings) { s.CapUSDDay = &five }) + }); err != nil { + t.Fatal(err) + } + oldOf, oldStamp := teamSpendOf, teamSpendStamp + var reads atomic.Int64 + var broken atomic.Bool + broken.Store(true) + teamSpendOf = func(_, _, _ string) (teams.Spend, error) { + reads.Add(1) + if broken.Load() { + return teams.Spend{}, errors.New("ledger unavailable") + } + return teams.Spend{}, nil + } + teamSpendStamp = func(_, _, _ string) string { return "same stamp" } + t.Cleanup(func() { teamSpendOf, teamSpendStamp = oldOf, oldStamp }) + manager := teamAgent(t, fixture, fixture.manager, oneAnswer(1), nil) + manager.teamBoundary() + want := "harbor has a $5 daily cap and today's spend could not be read (ledger unavailable), so nothing new starts until it can be read" + if got := manager.teamCapHold(fixture.profile, manager.teamRoles()); got != want { + t.Fatalf("unreadable spend held for %q, want %q", got, want) + } + if said, failed := callTool(t, manager.teamStartTool, `{"handle":"docs","brief":"write"}`); !failed || said != "No new member starts: "+want+"." { + t.Fatalf("team_start answered %q (failed %v)", said, failed) + } + webCalls := oneAnswer(1) + teamAgent(t, fixture, fixture.web, webCalls, nil) + appendTraffic(t, fixture, teams.Entry{Kind: teams.KindDirective, From: teams.FromManager, To: "web", Text: "Fix the header."}) + waitEvent(t, fixture, "held @web: "+want) + if webCalls.requests() != 0 { + t.Fatal("an unreadable ledger still woke the member") + } + if packets, err := teams.Packets(fixture.profile, teams.Person); err != nil || len(packets) != 0 { + t.Fatalf("an unreadable ledger raised a packet: %+v, %v", packets, err) + } + broken.Store(false) + if got := manager.teamCapHold(fixture.profile, manager.teamRoles()); got != "" || reads.Load() < 3 { + t.Fatalf("the next read was not retried: hold %q, reads %d", got, reads.Load()) + } + appendTraffic(t, fixture, teams.Entry{Kind: teams.KindDirective, From: teams.FromManager, To: "web", Text: "Try the header again."}) + waitRequests(t, webCalls, 1) +} + +func TestTeamCapHoldsWhenTheUsageLedgerPathIsADirectory(t *testing.T) { + t.Setenv("CODEAF_HOME", t.TempDir()) + fixture := newTeamFixture(t, true) + five := 5.0 + if err := teams.Update(fixture.profile, func(f *teams.File) error { + return f.SetSettings(fixture.teamID, func(s *teams.Settings) { s.CapUSDDay = &five }) + }); err != nil { + t.Fatal(err) + } + path := teams.UsageLedgerPath() + if err := os.MkdirAll(path, 0o700); err != nil { + t.Fatal(err) + } + manager := teamAgent(t, fixture, fixture.manager, nil, nil) + manager.teamBoundary() + if got := manager.teamCapHold(fixture.profile, manager.teamRoles()); !strings.Contains(got, "today's spend could not be read") { + t.Fatalf("a directory in place of the ledger held for %q", got) + } +} + +func TestTeamWithoutACapNeverReadsAnUnreadableLedger(t *testing.T) { + fixture := newTeamFixture(t, true) + old := teamSpendOf + reads := 0 + teamSpendOf = func(_, _, _ string) (teams.Spend, error) { + reads++ + return teams.Spend{}, errors.New("ledger unavailable") + } + t.Cleanup(func() { teamSpendOf = old }) + web := teamAgent(t, fixture, fixture.web, nil, nil) + web.teamBoundary() + if held := web.teamCapHold(fixture.profile, web.teamRoles()); held != "" || reads != 0 { + t.Fatalf("uncapped team held for %q after %d ledger reads", held, reads) + } +} + // WRAP UP FIRST ENDS IN A CLOSING REPORT, and the team closes when the person // accepts it. func TestTeamWrapUpEndsInAClosingReportThatClosesOnAccept(t *testing.T) { diff --git a/internal/session/team_questions.go b/internal/session/team_questions.go index abd1f8904..680b154ac 100644 --- a/internal/session/team_questions.go +++ b/internal/session/team_questions.go @@ -399,15 +399,19 @@ const teamPacketText = 3000 // team on a Close ([closingDecidedLine]). // - A MEMBER is told when its own question was answered (`◆ answered: …`) // or sent up. -func teamPacketLine(profile string, role teamRole, entry teams.Entry) string { +func teamPacketLine(profile string, role teamRole, entry teams.Entry) (string, string) { p, ok := packetOf(profile, entry) if !ok { - return "" + return "", "" } if closingDecided(role, entry, p) { - return closingDecidedLine(profile, p) + return closingDecidedLine(profile, p), p.ID + } + line := packetLineFor(role, entry, p) + if entry.State == teams.PacketDecided && line != "" && p.Kind != teams.PacketConflict { + return line, p.ID } - return packetLineFor(role, entry, p) + return line, "" } // packetOf is the packet a packet line is about, as it stands. diff --git a/internal/session/team_test.go b/internal/session/team_test.go index 6b1395345..59d195fbb 100644 --- a/internal/session/team_test.go +++ b/internal/session/team_test.go @@ -157,6 +157,99 @@ func TestTheManagerIsOfferedTheManagersVerbsAndNotThePost(t *testing.T) { } } +func TestTeamRoleDemotionRemovesOnlyLostVerbsAndRearmingAppends(t *testing.T) { + fixture := newTeamFixture(t, true) + manager := teamAgent(t, fixture, fixture.manager, nil, nil) + manager.teamBoundary() + before := manager.beltTools() + if err := teams.Update(fixture.profile, func(f *teams.File) error { + return f.SetManager(fixture.teamID, convKeyOf(t, fixture.parser)) + }); err != nil { + t.Fatal(err) + } + manager.teamBoundary() + for _, name := range teamToolNames[:8] { + if holds(manager, name) { + t.Errorf("demoted manager still holds %s", name) + } + } + if !holds(manager, teamPostToolName) || !holds(manager, teamRaiseToolName) { + t.Error("a demoted manager lost member verbs") + } + result := manager.executeTool(context.Background(), nil, nil, withdrawnCall("c1", teamSendToolName, `{}`), "") + if want := "team_send is no longer one of your tools: this conversation no longer manages a team."; result.text != want { + t.Errorf("removed manager verb answered %q, want %q", result.text, want) + } + kept := manager.beltTools() + j := 0 + for _, tool := range before { + for _, current := range kept { + if current.Name == tool.Name { + if kept[j].Name != tool.Name { + t.Fatal("surviving tools changed relative order") + } + j++ + break + } + } + } + if err := teams.Update(fixture.profile, func(f *teams.File) error { + return f.SetManager(fixture.teamID, convKeyOf(t, fixture.manager)) + }); err != nil { + t.Fatal(err) + } + manager.teamBoundary() + last := manager.beltTools() + if last[len(last)-1].Name != teamCloseReportToolName || !holds(manager, teamSendToolName) { + t.Fatal("manager verbs were not rearmed at the tail") + } + path := teams.Path(fixture.profile) + original, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, []byte("["), 0o600); err != nil { + t.Fatal(err) + } + manager.teamBoundary() + if !holds(manager, teamSendToolName) { + t.Fatal("an unreadable teams file removed the last known manager verbs") + } + if err := os.WriteFile(path, original, 0o600); err != nil { + t.Fatal(err) + } + if err := teams.Update(fixture.profile, func(f *teams.File) error { + return f.RemoveMember(fixture.teamID, convKeyOf(t, fixture.manager)) + }); err != nil { + t.Fatal(err) + } + manager.teamBoundary() + for _, name := range teamToolNames { + if holds(manager, name) { + t.Errorf("a conversation outside the team kept %s", name) + } + } + result = manager.executeTool(context.Background(), nil, nil, withdrawnCall("c2", teamPostToolName, `{}`), "") + if want := "team_post is no longer one of your tools: this conversation is no longer a member of a team with a manager."; result.text != want { + t.Errorf("removed member verb answered %q, want %q", result.text, want) + } +} + +func TestTeamToolsLeaveWhenTheTeamsFileIsGone(t *testing.T) { + fixture := newTeamFixture(t, true) + manager := teamAgent(t, fixture, fixture.manager, nil, nil) + manager.teamBoundary() + if err := os.Remove(teams.Path(fixture.profile)); err != nil { + t.Fatal(err) + } + manager.teamBoundary() + for _, name := range teamToolNames { + if holds(manager, name) { + t.Errorf("a missing teams file left %s on the belt", name) + } + } +} + func TestAMemberIsOfferedThePostAndNoManagersVerb(t *testing.T) { fixture := newTeamFixture(t, true) web := teamAgent(t, fixture, fixture.web, nil, nil) diff --git a/internal/session/team_wakewatch.go b/internal/session/team_wakewatch.go index 00ac96046..e3158f1f8 100644 --- a/internal/session/team_wakewatch.go +++ b/internal/session/team_wakewatch.go @@ -28,7 +28,8 @@ import ( // ([Agent.teamBoundary]). // - A MEMBER'S REPLY WAKES THE MANAGER: a team_post addressed to the manager, // and the events a member writes on its own (finished, failed, asking, -// teamevent.go). They COALESCE: the first one arms [teamWakeSettle], and one +// teamevent.go). A finished event after a reply in the same member turn +// starts no second wake. Other arrivals COALESCE: the first arms [teamWakeSettle], and one // turn carries everything that arrived by the time it runs out, because // three members finishing within a few seconds of each other is one thing // to act on, not three turns that each see a third of it. @@ -45,7 +46,9 @@ import ( // between conversations (team.go), so each conversation watches the log for // itself. It costs, per tick of [teamWatchEvery]: // -// - nothing at all while a turn is running, or once the session is closed; +// - while a turn runs, one stat of each managed member team's log, and a +// read only when it moves, so a manager's stop can reach a headless member; +// nothing once the session is closed; // - one stat of the teams file SHARED BY EVERY CONVERSATION IN THE PROCESS // ([sharedTeamsStamp]) while it is idle, which is how a conversation made a // member while it sat idle is noticed; @@ -104,6 +107,13 @@ const teamResumeWait = 60 * time.Second type teamWatch struct { cursors map[string]string trafficAt map[string]fileStamp + // stopCursors and stopAt are how far a RUNNING turn has looked for its + // manager's stop in each team's log ([Agent.teamStopTick]), and stopTurn + // is the turn they were taken for: a new turn starts again from the tail, + // because a stop written before it began is not addressed to it. + stopCursors map[string]string + stopAt map[string]fileStamp + stopTurn uint64 // pending is what has arrived for a manager since the first line that asks // for a wake, per team, and due is when the batch is handed over. pending map[string][]teams.Entry @@ -161,7 +171,7 @@ func (a *Agent) teamWatchLoop(profile string) { // teamWatchTick is one look, and false once the session has closed. func (a *Agent) teamWatchTick(profile string, now time.Time) bool { a.mu.Lock() - closed, running := a.closed, a.running + closed, running, turnAt, turnSerial := a.closed, a.running, a.teamTurnAt, a.teamTurnSerial a.mu.Unlock() if closed { return false @@ -170,6 +180,7 @@ func (a *Agent) teamWatchTick(profile string, now time.Time) bool { // costs nothing while no wrap-up is in progress. a.teamWrapUpDue(profile, now) if running { + a.teamStopTick(profile, turnAt, turnSerial) // A RUNNING TURN READS ITS OWN TRAFFIC at every step boundary. A batch a // manager was gathering goes with it: its posts land at the next // boundary, and its events are in the digest that turn carries. @@ -191,6 +202,10 @@ func (a *Agent) teamWatchTick(profile string, now time.Time) bool { arrived := a.teamWatchReadLocked(profile, role) var waking []teams.Entry for _, entry := range arrived { + if role.manager && entry.Kind == teams.KindEvent && entry.State == teams.StateFinished && + teamFinishedAfterReply(profile, role.id, entry) { + continue + } if teamWakes(role, entry) || teamPacketWakes(profile, role, entry) { waking = append(waking, entry) } @@ -224,6 +239,129 @@ func (a *Agent) teamWatchTick(profile string, now time.Time) bool { return true } +// teamStopTick is a running member performing its manager's stop itself. +// +// A STOP USED TO BE A WINDOW'S TO PERFORM, and only a window's: team_stop +// writes a [teams.KindStop] line and the interface holding the member ends its +// turn the way the person's Stop does (tui3's teamtraffic.go). A member codeaf +// opened in the background to run a woken turn ([rouseMember]) has no window, +// so its turn ran to its end whatever the manager said. The member's own +// session reads the same line instead, which works wherever it runs. +// +// IT COSTS one stat of each managed team's log per tick while a turn runs, and +// a read only when that log moved. Only a membership that takes this team's +// orders is looked at: not the manager's own, and not a shared one, whose +// manager here is a link and whose stops do not reach it (team_stop refuses +// them). A stop written before the running turn began is history, and a stop +// is performed only while the turn it was read in is still the one running. +func (a *Agent) teamStopTick(profile string, turnAt time.Time, serial uint64) { + a.team.mu.Lock() + roles := a.team.roles + if sharedTeamsStamp(profile) != a.team.teamsAt { + roles = a.teamRolesLocked(profile) + } + w := &a.team.watch + if w.stopCursors == nil || w.stopTurn != serial { + w.stopCursors, w.stopAt, w.stopTurn = map[string]string{}, map[string]fileStamp{}, serial + } + stop := false + for _, role := range roles { + if !role.managed || role.manager || role.shared || role.handle == "" { + continue + } + stamp := stampOf(teams.TrafficPath(profile, role.id)) + if stamp == w.stopAt[role.id] { + continue + } + entries, err := teams.ReadTraffic(profile, role.id, w.stopCursors[role.id], teamPageLimit) + if err != nil { + continue + } + for _, entry := range entries { + w.stopCursors[role.id] = entry.ID + if entry.Kind == teams.KindStop && entry.From == teams.FromManager && + (entry.Member == role.key || entry.Member == "" && entry.To == role.handle) && + !entry.At.Before(turnAt) { + stop = true + } + } + if len(entries) < teamPageLimit { + w.stopAt[role.id] = stamp + } else { + w.stopAt[role.id] = fileStamp{} + } + } + a.team.mu.Unlock() + if stop { + a.interruptTeamTurn(serial) + } +} + +// interruptTeamTurn is [Agent.Interrupt] with the manager's door, for the turn +// the stop was read in and no other. +// +// THE TURN IS ASKED FOR TWICE. A window holding the member may perform the same +// stop, and the turn it ends may be followed at once by another (a directive +// arriving behind the stop); a stop performed on that one would end work the +// manager never saw. So the serial is checked before anything is done, and +// again under the lock the cancel is taken under, and a turn that has ended +// is left alone, which is also what keeps a queued follow-up from being dropped +// by a stop that arrived after its turn was over. +func (a *Agent) interruptTeamTurn(serial uint64) { + live := func() bool { return a.running && a.teamTurnSerial == serial && a.cancel != nil } + a.mu.Lock() + ok := live() + a.mu.Unlock() + if !ok { + return + } + a.interruptDiscussions() + a.interrupt.begin() + a.mu.Lock() + if !live() { + a.mu.Unlock() + return + } + cancel := a.cancel + a.dropFollowUpsLocked() + a.stopSteerGraceLocked() + a.mu.Unlock() + cancel(stopFor(StopByManager)) +} + +// teamFinishedAfterReply reports whether a member's finished event ends a turn +// that already replied to the manager: scanning back from it through the same +// member's lines, a note to the manager comes before that member's previous +// turn ending (finished, failed, or stopped). +// +// ONE REPLY IS ONE WAKE. The reply woke the manager, or reached a turn it was +// already running; the ending of the same member turn arrives seconds later, +// often after the settle has handed the reply over, and waking again for it +// paid a second manager turn for news the first one already acted on, and +// tripped the loop breaker after half its rounds. It is read from the log and +// not remembered, because the manager may have restarted in between; and only +// when a finished event would otherwise wake or be handed over, which is rare. +// A failed or asking event still wakes: that is news the reply did not carry. +func teamFinishedAfterReply(profile, teamID string, finished teams.Entry) bool { + tail, err := teams.ReadTraffic(profile, teamID, "", teamFirstLook) + if err != nil { + return false + } + for i := len(tail) - 1; i >= 0; i-- { + e := tail[i] + if e.ID >= finished.ID || e.From != finished.From || (finished.Member != "" && e.Member != "" && e.Member != finished.Member) { + continue + } + if e.Kind == teams.KindEvent && (e.State == teams.StateFinished || e.State == teams.StateFailed || e.State == teams.StateIdle) { + return false + } + if e.Kind == teams.KindNote && e.To == teams.ToManager && strings.TrimSpace(e.Text) != "" { + return true + } + } + return false +} + // teamWatchReadLocked is what has been written to one team's log since this // conversation last looked, or nothing when the log has not moved. The caller // holds a.team.mu. @@ -269,7 +407,8 @@ func (a *Agent) teamWatchReadLocked(profile string, role teamRole) []teams.Entry // teamWakes reports whether an entry starts this conversation's turn when it // is idle: a directive from the manager to its handle or to everyone, for a // member; a member's post to the manager, or a member's finished, failed or -// asking event, for the manager. +// asking event, for the manager. The watch suppresses a finished event when +// the same turn already posted a reply. func teamWakes(role teamRole, entry teams.Entry) bool { if role.manager && teams.IsWrapUp(entry) { return true diff --git a/internal/session/team_wakewatch_test.go b/internal/session/team_wakewatch_test.go index 97dd87af6..fcd14b6c8 100644 --- a/internal/session/team_wakewatch_test.go +++ b/internal/session/team_wakewatch_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "strconv" "strings" "sync" "testing" @@ -205,6 +206,197 @@ func TestTeamWakeTheManagersWakeCoalescesABurst(t *testing.T) { } } +func TestTeamWakeAReplyAndItsFinishedTurnWakeTheManagerOnce(t *testing.T) { + fastTeamWake(t) + fixture := newWakingTeamFixture(t) + completer := oneAnswer(3) + manager := teamAgent(t, fixture, fixture.manager, completer, nil) + appendTraffic(t, fixture, teams.Entry{Kind: teams.KindNote, From: "web", To: teams.ToManager, Text: "The header is fixed."}) + waitRequests(t, completer, 1) + waitIdle(t, manager) + appendTraffic(t, fixture, teams.Entry{Kind: teams.KindEvent, From: "web", To: teams.ToManager, State: teams.StateFinished, Text: "finished"}) + time.Sleep(3*teamWakeSettle + 10*teamWatchEvery) + if got := trafficEvents(t, fixture, "woke ◆"); len(got) != 1 { + t.Fatalf("one reply and its ending woke the manager %d times", len(got)) + } + if got := trafficEvents(t, fixture, "finished"); len(got) != 1 { + t.Fatalf("the member's finished line was lost: %+v", got) + } + if got := completer.requests(); got != 1 { + t.Fatalf("the ending started %d extra model requests", got-1) + } + appendTraffic(t, fixture, teams.Entry{Kind: teams.KindEvent, From: "web", To: teams.ToManager, State: teams.StateFailed, Text: "failed: try again"}) + waitRequests(t, completer, 2) + if said := userTextIn(completer.request(1)); !strings.Contains(said, "@web finished its turn") { + t.Fatalf("the next manager turn did not receive the finished line:\n%s", said) + } + waitIdle(t, manager) + appendTraffic(t, fixture, teams.Entry{Kind: teams.KindEvent, From: "web", To: teams.ToManager, State: teams.StateAsking, Text: "asks: which file?"}) + waitRequests(t, completer, 3) +} + +func TestTeamWakeFinishedAfterReplyStillDeduplesAfterManagerRestart(t *testing.T) { + fastTeamWake(t) + fixture := newWakingTeamFixture(t) + firstCalls := oneAnswer(1) + manager := teamAgent(t, fixture, fixture.manager, firstCalls, nil) + appendTraffic(t, fixture, teams.Entry{Kind: teams.KindNote, From: "web", To: teams.ToManager, Text: "The header is fixed."}) + waitRequests(t, firstCalls, 1) + waitIdle(t, manager) + if err := manager.Close(); err != nil { + t.Fatal(err) + } + secondCalls := oneAnswer(1) + teamAgent(t, fixture, fixture.manager, secondCalls, nil) + appendTraffic(t, fixture, teams.Entry{Kind: teams.KindEvent, From: "web", To: teams.ToManager, State: teams.StateFinished, Text: "finished"}) + time.Sleep(3*teamWakeSettle + 10*teamWatchEvery) + if got := secondCalls.requests(); got != 0 { + t.Fatalf("the restarted manager made %d requests for the already answered ending", got) + } + if woke := trafficEvents(t, fixture, "woke ◆"); len(woke) != 1 { + t.Fatalf("the reply caused %d wakes across a restart", len(woke)) + } +} + +func TestTeamWakeTheLoopBreakerCountsTenMemberReplies(t *testing.T) { + fastTeamWake(t) + fixture := newWakingTeamFixture(t) + managerCalls := oneAnswer(teamLoopRounds + 1) + manager := teamAgent(t, fixture, fixture.manager, managerCalls, nil) + posted := make(chan int, teamLoopRounds+1) + releases := make([]chan struct{}, teamLoopRounds+1) + for i := range releases { + releases[i] = make(chan struct{}) + } + var web *Agent + // THE MEMBER'S ROUNDS ARE ARMED, NOT QUEUED. A positional script let any + // other request of the member's (an errand beside its turn, or a second + // look at a turn that had already replied) take the next round's step, post + // that round early and block the turn on a release nobody had given; under + // a long package run that is exactly what happened. Here only the request + // the test armed runs a round, and every other request is answered as done. + rounds := &armedRounds{armed: -1, run: func(ctx context.Context, i int) (*ai.Response, error) { + _, failed, err := web.teamPostTool(ctx, json.RawMessage(`{"to":"manager","text":"round `+strconv.Itoa(i)+` done"}`)) + if failed || err != nil { + return nil, errors.New("team_post failed") + } + posted <- i + select { + case <-releases[i]: + case <-ctx.Done(): + return nil, ctx.Err() + } + return textResponse("finished"), nil + }} + web = teamAgent(t, fixture, fixture.web, rounds, nil) + for i := 0; i <= teamLoopRounds; i++ { + rounds.arm(i) + events := mustSubmit(t, web, "answer the manager") + select { + case got := <-posted: + if got != i { + t.Fatalf("posted round %d before %d", got, i) + } + case <-time.After(5 * time.Second): + t.Fatal("the member did not post") + } + if i < teamLoopRounds { + waitRequests(t, managerCalls, i+1) + waitIdle(t, manager) + } else { + waitEvent(t, fixture, "the team has woken me 10 times") + } + close(releases[i]) + collect(t, events) + time.Sleep(3*teamWakeSettle + 10*teamWatchEvery) + if got := managerCalls.requests(); got != min(i+1, teamLoopRounds) { + t.Fatalf("after round %d the manager made %d calls", i, got) + } + } + if woke := trafficEvents(t, fixture, "woke ◆"); len(woke) != teamLoopRounds { + t.Fatalf("the team logged %d manager wakes, want %d", len(woke), teamLoopRounds) + } + log, err := teams.ReadTraffic(fixture.profile, fixture.teamID, "", 0) + if err != nil { + t.Fatal(err) + } + posts := 0 + for _, e := range log { + if e.Kind == teams.KindNote && e.From == "web" && e.To == teams.ToManager { + posts++ + } + } + if posts < teamLoopRounds { + t.Fatalf("the loop had only %d member posts", posts) + } +} + +func TestTeamStopEndsAHeadlessMembersCurrentTurn(t *testing.T) { + fastTeamWake(t) + fixture := newWakingTeamFixture(t) + manager := teamAgent(t, fixture, fixture.manager, nil, nil) + streaming := make(chan struct{}) + web := teamAgent(t, fixture, fixture.web, &scriptedCompleter{steps: []step{thinkingStep(streaming)}}, nil) + events := mustSubmit(t, web, "work on the header") + select { + case <-streaming: + case <-time.After(5 * time.Second): + t.Fatal("the member did not start") + } + said, failed := callTool(t, manager.teamStopTool, `{"handle":"web"}`) + if want := "Asked to stop @web's current turn. It ends the way the person's Stop does, whether a window has it open or codeaf opened it in the background: nothing is deleted, and its background tasks and jobs keep running."; failed || said != want { + t.Fatalf("team_stop answered %q (failed %v), want %q", said, failed, want) + } + collected := collect(t, events) + if last := collected[len(collected)-1]; last.Kind != EventTurnDone { + t.Fatalf("the member did not end after the manager's stop: %v", kinds(collected)) + } + if note, ok := firstOfKind(collected, EventNotice); !ok || note.Text != "the manager stopped this turn" { + t.Fatalf("the stopped turn said %+v (present %v)", note, ok) + } + if got := stopSentence(StopByManager, ""); got != "the manager stopped this turn" { + t.Fatalf("the manager stop sentence is %q", got) + } + if err := web.Close(); err != nil { + t.Fatal(err) + } + found := false + for _, row := range journaledErrors(t, fixture.web) { + found = found || strings.Contains(row.Message, string(StopByManager)) + } + if !found { + t.Fatal("the stopped turn's journal did not name the manager's door") + } +} + +func TestTeamStopIgnoresOldOtherAndSharedStops(t *testing.T) { + fastTeamWake(t) + fixture, sharedTeam, _ := linkedFixture(t) + started := make(chan struct{}) + web := teamAgent(t, fixture, fixture.web, &scriptedCompleter{steps: []step{thinkingStep(started)}}, nil) + // An old stop is history before this turn begins. + appendTraffic(t, fixture, teams.Entry{Kind: teams.KindStop, From: teams.FromManager, To: "web", Member: convKeyOf(t, fixture.web)}) + events := mustSubmit(t, web, "keep working") + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("the member did not start") + } + appendTraffic(t, fixture, teams.Entry{Kind: teams.KindStop, From: teams.FromManager, To: "parser", Member: convKeyOf(t, fixture.parser)}) + if err := teams.AppendTraffic(fixture.profile, sharedTeam, teams.Entry{Kind: teams.KindStop, From: teams.FromManager, To: "web", Member: convKeyOf(t, fixture.web)}); err != nil { + t.Fatal(err) + } + time.Sleep(10 * teamWatchEvery) + web.mu.Lock() + running := web.running + web.mu.Unlock() + if !running { + t.Fatal("an old, other-member, or shared stop ended this turn") + } + appendTraffic(t, fixture, teams.Entry{Kind: teams.KindStop, From: teams.FromManager, To: "web", Member: convKeyOf(t, fixture.web)}) + collect(t, events) +} + // A MEMBER'S EVENTS THAT ASK NOTHING WAKE NOBODY: a stop, a question coming // down, and the wake lines themselves. func TestTeamWakeTheManagerSleepsThroughLinesThatAskNothing(t *testing.T) { @@ -422,3 +614,34 @@ func TestTeamWakeAMemberThatCannotBeOpenedIsSaidInTheTraffic(t *testing.T) { t.Fatalf("a failed open was said %d times, want once", len(got)) } } + +// armedRounds is a member completer that runs one scripted round per request +// the test armed, whatever else the member asks in between: a namer, a +// narrator or a second look at its own turn is answered as done and never +// takes a round. +type armedRounds struct { + mu sync.Mutex + armed int + run func(ctx context.Context, round int) (*ai.Response, error) +} + +// arm makes the member's next turn request run round. +func (c *armedRounds) arm(round int) { + c.mu.Lock() + c.armed = round + c.mu.Unlock() +} + +func (c *armedRounds) CompleteWithMessages(ctx context.Context, messages []ai.Message, _ ...ai.Option) (*ai.Response, error) { + if isNameCall(messages) || isTitleCall(messages) || isCaptionCall(messages) { + return textResponse(""), nil + } + c.mu.Lock() + round := c.armed + c.armed = -1 + c.mu.Unlock() + if round < 0 { + return textResponse(checkpointNothingLeft), nil + } + return c.run(ctx, round) +} diff --git a/internal/session/team_wrapup.go b/internal/session/team_wrapup.go index 4ef73adee..02b3aad6f 100644 --- a/internal/session/team_wrapup.go +++ b/internal/session/team_wrapup.go @@ -217,8 +217,8 @@ func (a *Agent) teamWrapUpDue(profile string, now time.Time) { a.team.mu.Unlock() for _, w := range going { why := "" - spent := a.teamPoolSpend(profile, w.team, teamToday()) - if !w.measured { + spent, spendErr := a.teamPoolSpend(profile, w.team, teamToday()) + if !w.measured && spendErr == nil { a.team.mu.Lock() if held, ok := a.team.wraps[w.team]; ok { held.spentAt, held.measured = spent, true @@ -229,7 +229,7 @@ func (a *Agent) teamWrapUpDue(profile string, now time.Time) { switch { case now.Sub(w.started) >= wrapBound(w): why = fmt.Sprintf("the wrap-up ran out of time (%s) before the manager brought its report", wrapBound(w).Round(time.Minute)) - case spent-w.spentAt >= wrapUpSpendUSD: + case spendErr == nil && w.measured && spent-w.spentAt >= wrapUpSpendUSD: why = fmt.Sprintf("the wrap-up spent %s, its limit, before the manager brought its report", teamMoney(spent-w.spentAt)) default: continue @@ -341,7 +341,11 @@ func (a *Agent) teamCloseReportTool(ctx context.Context, args json.RawMessage) ( files = append(files, file) } } - spent := a.teamPoolSpend(profile, team.ID, teamToday()) + // A SPEND THAT CANNOT BE READ LEAVES THE FIGURE OUT of the report rather + // than holding the report back: the person is waiting on it to close the + // team, and a report drawn with no spend line is the emptiness law's + // unknown, where a refused report would leave the wrap-up to run out. + spent, _ := a.teamPoolSpend(profile, team.ID, teamToday()) raised, err := teams.Raise(profile, closingPacket(team.Name, team.ID, teams.ClosingReport{ Done: done, Left: strings.TrimSpace(parsed.Left), Files: files, SpendUSD: roundCents(spent), })) diff --git a/internal/session/tools_team.go b/internal/session/tools_team.go index 97609170d..09920c4ce 100644 --- a/internal/session/tools_team.go +++ b/internal/session/tools_team.go @@ -4,8 +4,8 @@ package session // // A team's manager is an ordinary conversation with every ordinary tool under // the ordinary approval rules (docs/design/conversations-and-teams/DESIGN.md, -// section 5). What makes it a manager is these five verbs and the digest its -// turns carry (team.go); what makes a member able to answer is the sixth. +// section 5). What makes it a manager is its verbs and the digest its turns +// carry (team.go); a member has the verbs to answer and raise a conflict. // // THEY ARE ONE GROUP, AND THE GROUP IS ARMED, NOT SHELVED. A manager needs its // verbs on the turn the person asks it to hand something out, so a round trip @@ -17,7 +17,7 @@ package session // goes through. The fixed prefix is unchanged for everybody else, which is what // prefixbudget_test.go holds. // -// SIX TOOLS AND NOT ONE WITH ACTIONS, for the reason the settings pair is two: +// SEPARATE TOOLS AND NOT ONE WITH ACTIONS, for the reason the settings pair is two: // THE APPROVAL GATE KEYS ON THE TOOL NAME. Reading a member's page and starting a // new conversation that spends money are two different acts, and a person must // be able to allow one and be asked about the other. So the reads, the messages @@ -26,7 +26,8 @@ package session // // THE CHANNEL IS THE TRAFFIC LOG AND NOTHING ELSE. Every write here is one // [teams.AppendTraffic]: a message is a note or a directive, a stop is a -// [teams.KindStop] entry the interface performs as the person's own Stop, and a +// [teams.KindStop] entry the member's engine watches while its turn runs, and +// a window holding it may also perform as the person's own Stop; a // start is a [teams.KindStart] entry the interface performs by opening the new // conversation, and the new conversation reads its brief off that same entry // (teamevent.go's [teamBriefLine]), marked as the manager's and never the person's. None of these verbs reaches into another @@ -161,9 +162,8 @@ func (a *Agent) memberTools() []bare.Tool { // refusal to hand the model when there is none. // // IT IS ASKED ON EVERY CALL, never remembered from the boundary that armed the -// verb: a conversation removed as manager a minute ago keeps the verb on its -// belt (team.go states why), and this is where it is told it no longer runs -// that team. +// verb: a role can change between a boundary and a call. This is where a +// still-running call learns it no longer has that role. func (a *Agent) teamTarget(want string, manager bool) (teams.Team, teamRole, string) { profile := a.config.teamProfile() if profile == "" { @@ -621,7 +621,7 @@ func (a *Agent) teamStopTool(ctx context.Context, args json.RawMessage) (string, if err := teams.AppendTraffic(a.config.teamProfile(), team.ID, entry); err != nil { return "The stop could not be written to the team's traffic: " + err.Error(), true, nil } - return fmt.Sprintf("Asked to stop @%s's current turn. A window that has it open ends the turn the way the person's Stop does. A member codeaf opened in the background, with no window on it, is not stopped: its turn runs to its end.", member.Handle), false, nil + return fmt.Sprintf("Asked to stop @%s's current turn. It ends the way the person's Stop does, whether a window has it open or codeaf opened it in the background: nothing is deleted, and its background tasks and jobs keep running.", member.Handle), false, nil } // ── team_start ────────────────────────────────────────────────────────────── diff --git a/internal/teams/decision.go b/internal/teams/decision.go index 7b58ced36..1c5e6435d 100644 --- a/internal/teams/decision.go +++ b/internal/teams/decision.go @@ -37,8 +37,8 @@ import ( // its members raised. // // THE FILE IS EVENTS, FOLDED BY ID. One JSON object per line, only ever -// appended to: a `raise` carrying the whole packet, then `decide` or -// `escalate` lines naming it by id. Reading folds them in file order into the +// appended to: a `raise` carrying the whole packet, then `decide`, +// `escalate`, or `told` lines naming it by id. Reading folds them in file order into the // packet as it stands now. Nothing is rewritten, so a reader never sees a // half-written packet and a writer never loses another's line. // @@ -59,11 +59,11 @@ import ( // THE FILE ROTATES LIKE TRAFFIC, and keeps every packet still waiting. Past // [decisionsRotateBytes] the file is renamed to decisions.1.jsonl (replacing // the one before) and the new file opens with one `carry` line per packet -// still waiting, the packet whole as it stands (its trail, its escalated -// state). A reader folds the rotated file and then the current one, and a -// carry replaces what the rotated file said of that id, so a waiting packet -// is never lost to a rotation and a decided one stays readable for one -// rotation more, which is long enough for its raiser to be handed the answer. +// still waiting, every packet for today's cap, and the newest decided answers +// still owed to their raisers within half the rotation size. A reader folds +// the rotated file and then the current one; a carry replaces the older +// state. Oldest owed answers beyond that bound remain in the rotated file +// for one more rotation. // // EVERY CHANGE IS ALSO A LINE OF TRAFFIC, a [KindPacket] entry in the log of // the team that raised it, of every team that was asked to decide it, and of @@ -206,6 +206,8 @@ type Packet struct { // Decision is the chosen option's id, or the person's own words when none // fitted. Decision string `json:"decision,omitempty"` + // Told records that the raiser was handed a decided answer. + Told bool `json:"told,omitempty"` // Reason is the decider's reason, or the last escalation's. Reason string `json:"reason,omitempty"` // Trail is every escalation, oldest first. @@ -238,7 +240,8 @@ var ( // packet on the event has been replaced with the one that was there, and // nothing was written. It stays inside this package: [Raise] answers that // packet and no error. - errCapAlready = errors.New("teams: that cap crossing was already raised") + errCapAlready = errors.New("teams: that cap crossing was already raised") + errAlreadyTold = errors.New("teams: that answer was already handed over") ) // decisionEvent is one line of a packet file. @@ -257,7 +260,8 @@ const ( opRaise = "raise" opDecide = "decide" opEscalate = "escalate" - // opCarry is a waiting packet written again, whole, at the head of a + opTold = "told" + // opCarry is a retained packet written again, whole, at the head of a // rotated file. It replaces what an older line said of that id. opCarry = "carry" ) @@ -406,6 +410,29 @@ func Decide(profileDir, id, by, decision, reason string) (Packet, error) { return out, nil } +// Told records that a decided packet's answer reached its raiser. A duplicate +// delivery changes nothing, and the append uses the packet file's own lock. +func Told(profileDir, id string) error { + origin, err := packetOrigin(profileDir, id) + if err != nil { + return err + } + e := decisionEvent{Op: opTold, At: time.Now(), ID: id} + err = appendDecision(profileDir, origin, &e, func(p Packet) error { + if p.Told { + return errAlreadyTold + } + if p.State != PacketDecided { + return ErrNoPacket + } + return nil + }) + if errors.Is(err, errAlreadyTold) { + return nil + } + return err +} + // Escalate sends packet id up: to an open team above the one it waits on, or // to [Person]. by is as for [Decide]. Down, sideways or to a closed team is // [ErrSideways]; a decided packet is [ErrDecided]. @@ -662,23 +689,38 @@ func appendDecision(profileDir, teamID string, e *decisionEvent, check func(Pack // rotateDecisions starts a new packet file at path, under its lock: the fold // as it stands is taken, the file is renamed over the rotated one, and the new -// file is written (temporary file and rename) with a carry line for every -// packet still waiting, so nothing waiting lives only in the rotated file. +// file is written (temporary file and rename) with waiting packets, today's +// cap packets, and bounded owed answers. Newer answers take the available +// decided space first; older ones remain in the rotated file for one pass. func rotateDecisions(path string) error { now, err := packetCache.read(path) if err != nil { return err } var carry bytes.Buffer + var owed []Packet + today := Today() for _, p := range now.list() { - if !p.Waiting() { - continue + switch { + case p.Waiting(), p.Kind == PacketCap && p.Cap != nil && p.Cap.Day == today: + if err := writeDecisionCarry(&carry, p); err != nil { + return err + } + case p.State == PacketDecided && !p.Told && p.Kind != PacketCap && p.Kind != PacketConflict: + owed = append(owed, p) } - p := p - line, err := json.Marshal(decisionEvent{Op: opCarry, At: p.At, Packet: &p}) + } + sort.SliceStable(owed, func(i, j int) bool { return owed[i].At.After(owed[j].At) }) + decidedBytes := int64(0) + for _, p := range owed { + line, err := decisionCarryLine(p) if err != nil { return err } + if decidedBytes+int64(len(line)+1) >= decisionsRotateBytes/2 { + continue + } + decidedBytes += int64(len(line) + 1) carry.Write(line) carry.WriteByte('\n') } @@ -705,6 +747,20 @@ func rotateDecisions(path string) error { return os.Rename(name, path) } +func decisionCarryLine(p Packet) ([]byte, error) { + return json.Marshal(decisionEvent{Op: opCarry, At: p.At, Packet: &p}) +} + +func writeDecisionCarry(dst *bytes.Buffer, p Packet) error { + line, err := decisionCarryLine(p) + if err != nil { + return err + } + dst.Write(line) + dst.WriteByte('\n') + return nil +} + // logPacket appends a [KindPacket] line to the Traffic of each distinct team // in teams that is a team (not [Person], not ""). A log that cannot be written // costs the line and never the packet, which is written already. @@ -872,6 +928,11 @@ func (f *folded) apply(e decisionEvent) { if p, ok := f.byID[e.ID]; ok && p.Waiting() { f.byID[e.ID] = foldEscalate(p, e) } + case opTold: + if p, ok := f.byID[e.ID]; ok && p.State == PacketDecided { + p.Told = true + f.byID[e.ID] = p + } } } diff --git a/internal/teams/wrapup_test.go b/internal/teams/wrapup_test.go index 5723664a3..b28ed3d96 100644 --- a/internal/teams/wrapup_test.go +++ b/internal/teams/wrapup_test.go @@ -1,6 +1,8 @@ package teams import ( + "bufio" + "encoding/json" "os" "strings" "testing" @@ -68,6 +70,102 @@ func TestThePacketFileRotatesAndKeepsEveryWaitingPacket(t *testing.T) { } } +func TestRotationKeepsAnUntoldAnswerAcrossSeveralFiles(t *testing.T) { + dir := packetTeams(t) + old := decisionsRotateBytes + decisionsRotateBytes = 3000 + t.Cleanup(func() { decisionsRotateBytes = old }) + owed, err := Raise(dir, Packet{Team: "bbbbbbbbbbbb", Kind: PacketQuestion, RaisedBy: "web", Question: "JSON or form data?"}) + must(t, err) + _, err = Decide(dir, owed.ID, FromManager, "JSON", "the other endpoints use it") + must(t, err) + var firstConflict string + for i := 0; i < 24; i++ { + p, err := Raise(dir, conflict()) + must(t, err) + if i == 0 { + firstConflict = p.ID + } + _, err = Decide(dir, p.ID, "lead", "1", "matches") + must(t, err) + } + if _, err := PacketByID(dir, firstConflict); err != ErrNoPacket { + t.Fatalf("an old conflict was carried as an owed answer: %v", err) + } + if got, err := PacketByID(dir, owed.ID); err != nil || got.State != PacketDecided { + t.Fatalf("an untold answer was lost after rotations: %+v, %v", got, err) + } + if err := Told(dir, owed.ID); err != nil { + t.Fatal(err) + } + if got, err := PacketByID(dir, owed.ID); err != nil || !got.Told { + t.Fatalf("the delivery mark was not folded: %+v, %v", got, err) + } + for i := 0; i < 24; i++ { + p, err := Raise(dir, conflict()) + must(t, err) + _, err = Decide(dir, p.ID, "lead", "1", "matches") + must(t, err) + } + if got, err := PacketByID(dir, owed.ID); err != ErrNoPacket { + t.Fatalf("a told answer remains after two more rotations: %+v, %v", got, err) + } +} + +func TestTodaysCapDecisionSurvivesRotationAndOwedCarryStaysBounded(t *testing.T) { + dir := packetTeams(t) + old := decisionsRotateBytes + decisionsRotateBytes = 3000 + t.Cleanup(func() { decisionsRotateBytes = old }) + cap, err := Raise(dir, capAsk(Today(), 5)) + must(t, err) + _, err = Decide(dir, cap.ID, Person, OptionRaiseCap, "carry on") + must(t, err) + var newest Packet + for i := 0; i < 4; i++ { + newest, err = Raise(dir, Packet{Team: "bbbbbbbbbbbb", Kind: PacketQuestion, RaisedBy: "web", + Question: strings.Repeat("which way now? ", 25)}) + must(t, err) + _, err = Decide(dir, newest.ID, FromManager, "go", "because") + must(t, err) + } + for i := 0; i < 24; i++ { + p, err := Raise(dir, conflict()) + must(t, err) + _, err = Decide(dir, p.ID, "lead", "1", "matches") + must(t, err) + } + if got, err := PacketByID(dir, cap.ID); err != nil || got.Decision != OptionRaiseCap || got.Cap.RaiseTo != 10 { + t.Fatalf("today's cap decision disappeared: %+v, %v", got, err) + } + if again, err := Raise(dir, capAsk(Today(), 5)); err != nil || again.ID != cap.ID { + t.Fatalf("the same crossing raised again: %+v, %v", again, err) + } + if got, err := PacketByID(dir, newest.ID); err != nil || got.State != PacketDecided { + t.Fatalf("the newest owed answer was not carried: %+v, %v", got, err) + } + file, err := os.Open(DecisionsPath(dir, "bbbbbbbbbbbb")) + must(t, err) + defer file.Close() + scan := bufio.NewScanner(file) + decidedCarry := int64(0) + for scan.Scan() { + var line decisionEvent + if err := json.Unmarshal(scan.Bytes(), &line); err != nil { + t.Fatal(err) + } + if line.Op == opCarry && line.Packet != nil && line.Packet.State == PacketDecided && line.Packet.Kind != PacketCap { + decidedCarry += int64(len(scan.Bytes()) + 1) + } + } + if err := scan.Err(); err != nil { + t.Fatal(err) + } + if decidedCarry == 0 || decidedCarry >= decisionsRotateBytes/2 { + t.Fatalf("decided carry used %d bytes, want some and no more than half of %d", decidedCarry, decisionsRotateBytes) + } +} + // A QUESTION'S ANSWER READS AS ONE ON THE RAIL. func TestAQuestionsAnswerIsLoggedAsAnswered(t *testing.T) { dir := packetTeams(t) diff --git a/internal/tui3/teamclose.go b/internal/tui3/teamclose.go index 58ee63b2f..30761f596 100644 --- a/internal/tui3/teamclose.go +++ b/internal/tui3/teamclose.go @@ -1,6 +1,7 @@ package tui3 import ( + "sort" "strings" "time" @@ -26,8 +27,13 @@ import ( // Closed, in one step, with Undo. // - `Cancel`. // -// A CONVERSATION THAT IS ALSO IN ANOTHER OPEN TEAM IS NEVER STOPPED OR CLOSED -// by this: it is still that team's. Closing a team closes the teams under it +// A CONVERSATION THAT IS ALSO IN ANOTHER OPEN TEAM IS NEVER CLOSED by this, and +// never stopped, with ONE EXCEPTION: the manager of a team being closed. A +// sub-team's manager is also a member of the team above, so it is still that +// team's and keeps its tab; but the turn it is running is this team's work, so +// it counts as working (the card, and `Wrap up first`, rather than a close at +// once while it runs on) and `Close now` stops that turn. Closing a team +// closes the teams under it // (internal/teams' [teamstore.File.Close]); reopening reopens exactly what the // close closed. Delete is only ever offered on a closed team, and forgets the // grouping, the Traffic and the packets; the conversations stay in history. @@ -80,6 +86,24 @@ func (a *app) teamsCloseKeys(id string) []string { return keys } +// teamsClosingManagers includes the manager of each team being closed even +// when that conversation is also a member of an open team above it. +func (a *app) teamsClosingManagers(id string) map[string]bool { + closing := map[string]bool{id: true} + for _, t := range a.teamTree().Descendants(id) { + if !t.Closed() { + closing[t.ID] = true + } + } + managers := map[string]bool{} + for _, t := range a.wall.teams { + if closing[t.ID] && t.Manager != "" { + managers[t.Manager] = true + } + } + return managers +} + // teamsRunning is the members of team id (and the teams under it) that are // working now, by handle or name, and whether its manager is one of them or // has any running at all. @@ -89,11 +113,30 @@ func (a *app) teamsRunning(id string) (names []string, managed bool) { return nil, false } front := a.frontTabKey() - for _, key := range a.teamsCloseKeys(id) { + managers := a.teamsClosingManagers(id) + keys := a.teamsCloseKeys(id) + seen := map[string]bool{} + for _, key := range keys { + seen[key] = true + } + var extra []string + for key := range managers { + if !seen[key] { + extra = append(extra, key) + } + } + sort.Strings(extra) + keys = append(keys, extra...) + for _, key := range keys { if !a.trafficHeld(key) || a.tabSignalFor(key, key == front) != tabWorking { continue } name := key + if managers[key] { + name = a.teamManagerMark() + " manager" + names = append(names, name) + continue + } for _, u := range a.wall.teams { if m, ok := u.Member(key); ok { name = m.Word @@ -160,7 +203,10 @@ func (a *app) teamsCloseNow(id, report string) tea.Cmd { func (a *app) teamsStopMembers(id string) []string { front := a.frontTabKey() var shut []string - for _, key := range a.teamsCloseKeys(id) { + closing := a.teamsCloseKeys(id) + covered := map[string]bool{} + for _, key := range closing { + covered[key] = true if !a.trafficHeld(key) { continue } @@ -176,6 +222,19 @@ func (a *app) teamsStopMembers(id string) []string { shut = append(shut, key) } } + var extra []string + for key := range a.teamsClosingManagers(id) { + extra = append(extra, key) + } + sort.Strings(extra) + for _, key := range extra { + if covered[key] || !a.trafficHeld(key) || a.tabSignalFor(key, key == front) != tabWorking { + continue + } + // A manager shared with an open team keeps its tab but cannot keep + // working on the team being closed. + a.trafficStop(key) + } return shut } diff --git a/internal/tui3/teamspage_test.go b/internal/tui3/teamspage_test.go index 9b8a74120..3cc6678ca 100644 --- a/internal/tui3/teamspage_test.go +++ b/internal/tui3/teamspage_test.go @@ -359,6 +359,43 @@ func TestTeamsCloseWithNothingRunningIsOneClickAndUndo(t *testing.T) { } } +func TestTeamsCloseCountsAndStopsASubteamManagerSharedWithItsParent(t *testing.T) { + a, harbor, orbit := teamsPlaceLabIDs(t) + var shared chatTab + for _, tab := range a.tabList() { + if tab.key != a.frontTabKey() { + if parent, ok := a.teamByID(harbor); ok && parent.Holds(tab.key) { + shared = tab + break + } + } + } + if shared.key == "" { + t.Fatal("no shared member in the fixture") + } + if err := a.teamAdd(orbit, []chatTab{shared}); err != nil { + t.Fatal(err) + } + if err := a.teamMakeManager(orbit, shared); err != nil { + t.Fatal(err) + } + member := &fakeAgent{model: "m"} + watch := &behindWatch{} + watch.turning.Store(true) + a.behind[shared.key] = &kept{conv: Conversation{Agent: member, SessionFile: shared.file}, watch: watch} + drive(t, a, runCmd(a.teamsCloseAsk(orbit))...) + if !a.tsheet.on || a.tsheet.cursor != tsWrapUp { + t.Fatalf("the working sub-team manager was skipped: %+v", a.tsheet) + } + if text := teamsFrameText(a); !strings.Contains(text, a.teamManagerMark()+" manager") { + t.Fatalf("the card does not name the working manager:\n%s", text) + } + drive(t, a, runCmd(a.teamSheetDo(tsCloseNow))...) + if member.stops != 1 || a.tabShut[shared.key] { + t.Fatalf("Close now stopped %d turns and shut the parent's tab=%v", member.stops, a.tabShut[shared.key]) + } +} + // SOMETHING RUNNING PUTS UP THE CARD: `Close now` first when no manager runs // the team, `Wrap up first` first when one does, and Cancel changes nothing. func TestTeamsCloseCardOffersWrapUpNowAndCancel(t *testing.T) {