pool

func (*Pool) Cancel ¶

func (p *Pool) Cancel(receiptID string) error

Cancel cancels the child running for receiptID — the one explicit cancellation an async child has (D4). The child's session records the canceled turn and the receipt settles canceled; a receipt that is queued but not yet running is dequeued the same way. A receipt with no running child — settled, parked at an approval, or never this pool's — fails with ErrNotRunning: a parked child's fate is the decision that resumes it, not a cancellation.

func (*Pool) Close ¶

func (p *Pool) Close(ctx context.Context) error

Close stops taking work and drains: every running child is canceled — async ones on their pool-owned contexts, sync ones through their registered cancels, their turns settling with their parents — and Close waits for the pool's own goroutines to settle their receipts, every child's session recording its final state before the process gives up the storage (DeerFlow's gateway drain). ctx bounds the wait; its error returns with the drain still progressing. A second Close returns nil — the drain is already done or underway.

func (*Pool) Decide ¶

func (p *Pool) Decide(ctx context.Context, parent *thread.Session, ds ...thread.Decision) (*thread.Turn, error)

Decide records decisions over a parent session's nested approvals and resumes the children they address (ADR 0022 §7): decisions are recorded in the parent — validation, audit and the signed challenge all live there — then the pool replays them into the child session, which resumes exactly as a top-level session does (ADR 0021 §1). When the child runs to its end the delegation completes: a sync one's parked wrapper call resolves with the child's answer (the core's Resolve — re-running the delegation would replay the child), an async one's receipt settles. A child that parks again on the way mirrors its new requests and waits for the next Decide.

With no decisions it is a pure pump: children whose every mirrored request now has an effective decision — or has lapsed past its expiry — resume, which is how a signed DecideSigned on the session drives children too.

The returned Turn is the resumed child's, when the call resumed one; the parent's own continuation is reached through its parked boundary as always. The wrapper call never takes a direct decision: Pending shows the child's requests, and the wrapper completes through them.

package main

import (
	"context"
	"fmt"

	"github.com/weftgo/weft"
	"github.com/weftgo/weft/thread"
	"github.com/weftgo/weft/thread/pool"
	"github.com/weftgo/weft/wefttest"
)

func main() {
	ctx := context.Background()
	gated := weft.Tool("refund", "", func(_ context.Context, in struct{ OrderID string }) (string, error) {
		return "refunded " + in.OrderID, nil
	}, weft.RequireApproval())
	child := weft.New(wefttest.Script(
		wefttest.ToolCalls(wefttest.Call{Name: "refund", Args: `{"order_id":"42"}`}),
		wefttest.Say("refund issued"),
	), gated)
	p := pool.New(2)
	parent := weft.New(wefttest.Script(
		wefttest.ToolCalls(wefttest.Call{Name: "research",
			Args: wefttest.Args(struct{ Prompt string }{"refund order 42"})}),
		wefttest.Say("handled"),
	), p.Wrap("research", "delegates the refund flow", child))
	s, _ := thread.Create(ctx, thread.Memory(), parent)
	t1, _ := s.Send(ctx, weft.User("refund order 42"))
	if _, err := t1.Wait(); err != nil {
		panic(err)
	}
	pend := s.Pending()
	fmt.Println("pending:", pend[0].Tool)
	childTurn, err := p.Decide(ctx, s, thread.Approve(pend[0].CallID))
	if err != nil {
		panic(err)
	}
	if _, err := childTurn.Wait(); err != nil {
		panic(err)
	}
	res, err := t1.Next().Wait() // the parent's parked run, resumed with the answer
	if err != nil {
		panic(err)
	}
	fmt.Println("parent:", res.Text())
	for _, r := range pool.Receipts(s) {
		fmt.Println("receipt:", r.State, "-", r.Stop)
	}
}

Output:
pending: refund
parent: handled
receipt: done - refund issued

func (*Pool) Forward ¶

func (p *Pool) Forward(ctx context.Context, receiptID string, msg weft.Message) (*thread.Turn, error)

Forward steers a running child with msg, explicitly (ADR 0022 §8, ADR 0019 §7): the message is sent to the child's session under the Steer policy — delivered at the run's next steering drain point, or deferred to a follow-up turn when the run ends on an intended stop or an open approval boundary; the child's session records the receipt either way. Nothing is ever forwarded implicitly. The returned Turn is the steer's receipt turn; a child that is not running (queued, parked awaiting decisions, settled) refuses with ErrNotRunning.

func (*Pool) Register ¶

func (p *Pool) Register(sessionID string, agent *weft.Agent)

Register re-attaches an agent to a child session the process no longer holds one for — the restart hook (ADR 0022 §7): a parked child's boundary outlives the process, and resuming it needs the agent that runs it, which no file carries. Children a Wrap made need no call: the wrap's name, recorded in the child's header metadata, is looked up among the pool's wrapped agents. Register is for the rest — Submit children — and for wraps renamed between runs. A nil agent is ignored.

func (*Pool) Submit ¶

func (p *Pool) Submit(ctx context.Context, parent *thread.Session, agent *weft.Agent, prompt string) (*Receipt, error)

Submit hands a task to a child session of parent and returns at once (ADR 0022 §2): the child session is created in the parent's storage with a lineage naming the parent, the acceptance receipt is durable in the parent session before Submit returns, and the child runs on a pool-owned context — independent of the submitting caller's cancellation by design (D4), canceled explicitly by Cancel or Close. agent must not be nil; the receipt's Child names the session to watch (the v0.4 Watcher) or reopen.

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/weftgo/weft"
	"github.com/weftgo/weft/thread"
	"github.com/weftgo/weft/thread/pool"
	"github.com/weftgo/weft/wefttest"
)

func main() {
	ctx := context.Background()
	s, _ := thread.Create(ctx, thread.Memory(), weft.New(wefttest.Script()))
	p := pool.New(1)
	child := weft.New(wefttest.Script(wefttest.Say("background answer")))
	r, err := p.Submit(ctx, s, child, "work in the background")
	if err != nil {
		panic(err)
	}
	fmt.Println("accepted:", r.State)
	// The child settles on the pool's goroutine; a live application
	// reads the receipt back — polled here, bounded for the example's
	// sake — and closes the pool only to drain what is still flying.
	var final pool.Receipt
	for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); {
		for _, rc := range pool.Receipts(s) {
			if rc.ID == r.ID {
				final = rc
			}
		}
		if final.State == thread.PoolDone {
			break
		}
		time.Sleep(time.Millisecond)
	}
	if err := p.Close(ctx); err != nil {
		panic(err)
	}
	fmt.Println("settled:", final.State, "-", final.Stop)
}

Output:
accepted: accepted
settled: done - background answer

func (*Pool) Wrap ¶

func (p *Pool) Wrap(name, description string, agent *weft.Agent, opts ...WrapOption) *weft.ToolDef

Wrap returns a subagent tool whose child runs hold one pool slot (ADR 0022 §1–§2): a delegation tool over agent, built on the core's Subagent, run by the pool as a child session of the session whose run is calling it — linked by lineage, receipted in the parent, and bounded by the semaphore.

Inside a session's run (the parent found through the run context) the child is a session of its own in the parent's storage: sync by default — the call waits and the result is the child's answer; with Async the result is the receipt and the child runs on. A wrapped tool called outside any session — a bare Generate — has no parent to receipt into and falls back to the ordinary subagent path under the slot, ADR 0014's semantics unchanged.

A delegation chain deeper than the pool's max is refused with the core's SUBAGENT_CYCLE code before any child starts: every sync level holds a slot while it waits, so a chain of max+1 could only deadlock — the pool's bound doubles as the depth guard the core deliberately left unset (ADR 0014).

package main

import (
	"context"
	"fmt"

	"github.com/weftgo/weft"
	"github.com/weftgo/weft/thread"
	"github.com/weftgo/weft/thread/pool"
	"github.com/weftgo/weft/wefttest"
)

func main() {
	ctx := context.Background()
	st := thread.Memory()
	child := weft.New(wefttest.Script(wefttest.Say("the answer is 42")))
	p := pool.New(2)
	parent := weft.New(wefttest.Script(
		wefttest.ToolCalls(wefttest.Call{Name: "research",
			Args: wefttest.Args(struct{ Prompt string }{"find the answer"})}),
		wefttest.Say("done"),
	), p.Wrap("research", "delegates research to a child session", child))
	s, _ := thread.Create(ctx, st, parent)
	turn, err := s.Send(ctx, weft.User("what is the answer?"))
	if err != nil {
		panic(err)
	}
	res, err := turn.Wait()
	if err != nil {
		panic(err)
	}
	fmt.Println("reply:", res.Text())
	for _, r := range pool.Receipts(s) {
		fmt.Println("receipt:", r.State, "-", r.Stop)
	}
	fmt.Println("delegated output tokens:", s.Usage().Delegated.OutputTokens)
}

Output:
reply: done
receipt: done - the answer is 42
delegated output tokens: 5