Skip to content

Commit 6e5a8b1

Browse files
ksarangmathclaude
andcommitted
perf: lock the parent up front for predictable parent effects
Roots whose transition will report to the parent now plan the agent family before writing: a model output that ends the turn without tool calls, a terminal model failure and a terminal compaction failure. These complete in one attempt instead of rolling back to add the parent. Outputs that call tools still lock only the child. AcceptOutput and the planning use one shared rule for which outputs report to the parent. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
1 parent bfe570e commit 6e5a8b1

21 files changed

Lines changed: 214 additions & 40 deletions

‎internal/storage/executionstore/agent_inputs_backlog_store.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,7 @@ func (s *Store) changeBacklog(ctx context.Context, projectID, agentID uuid.UUID,
5353
if projectID == uuid.Nil || agentID == uuid.Nil || change.ID == uuid.Nil {
5454
return struct{}{}, errors.New("project, agent and input are required")
5555
}
56-
unit, h, err := beginExecution(ctx, s, projectID, agentID)
56+
unit, h, err := beginExecution(ctx, s, projectID, agentID, false)
5757
if err != nil {
5858
return struct{}{}, err
5959
}

‎internal/storage/executionstore/agent_interactions_store.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -81,7 +81,7 @@ func (s *Store) CreatePermissionInteraction(
8181
"CreatePermissionInteraction",
8282
func(ctx context.Context) (AgentInteractionRecord,
8383
error) {
84-
unit, h, err := beginExecution(ctx, s, input.ProjectID, input.AgentID)
84+
unit, h, err := beginExecution(ctx, s, input.ProjectID, input.AgentID, false)
8585
if err != nil {
8686
return AgentInteractionRecord{}, err
8787
}

‎internal/storage/executionstore/agent_runtime_cancellation_store.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ func (s *Store) CancelAgent(ctx context.Context, input CancelAgentInput) (Cancel
2828
return agentexecution.Retry(ctx,
2929
"CancelAgent",
3030
func(ctx context.Context) (CancelAgentResult, error) {
31-
unit, h, err := beginExecution(ctx, s, input.ProjectID, input.AgentID)
31+
unit, h, err := beginExecution(ctx, s, input.ProjectID, input.AgentID, false)
3232
if err != nil {
3333
return CancelAgentResult{}, err
3434
}

‎internal/storage/executionstore/agent_runtime_locks_store.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -54,7 +54,7 @@ func (s *Store) EnsureRuntimeLockActive(
5454
if projectID == uuid.Nil || agentID == uuid.Nil || runtimeID == uuid.Nil {
5555
return struct{}{}, errors.New("project, agent, and runtime lock ids are required")
5656
}
57-
unit, h, err := beginExecution(ctx, s, projectID, agentID)
57+
unit, h, err := beginExecution(ctx, s, projectID, agentID, false)
5858
if errors.Is(err, storeerr.ErrNotFound) {
5959
return struct{}{}, storeerr.ErrRuntimeLockInactive
6060
}
@@ -119,7 +119,7 @@ func (s *Store) ReleaseAgentRuntimeLock(
119119
_, err := agentexecution.Retry(ctx,
120120
"ReleaseAgentRuntimeLock",
121121
func(ctx context.Context) (struct{}, error) {
122-
unit, h, err := beginExecution(ctx, s, projectID, agentID)
122+
unit, h, err := beginExecution(ctx, s, projectID, agentID, false)
123123
if errors.Is(err, storeerr.ErrNotFound) {
124124
return struct{}{}, storeerr.ErrRuntimeLockInactive
125125
}

‎internal/storage/executionstore/agent_work_store.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ func (s *Store) AdvanceOwnedAgentWork(
3333
if input.ProjectID == uuid.Nil || input.AgentID == uuid.Nil || input.RuntimeLockID == uuid.Nil {
3434
return executionResult{ClaimedAgentWork{}, false}, errors.New("project, agent, and runtime lock ids are required")
3535
}
36-
unit, h, err := beginExecution(ctx, s, input.ProjectID, input.AgentID)
36+
unit, h, err := beginExecution(ctx, s, input.ProjectID, input.AgentID, false)
3737
if errors.Is(err, storeerr.ErrNotFound) {
3838
return executionResult{ClaimedAgentWork{}, false}, storeerr.ErrRuntimeLockInactive
3939
}

‎internal/storage/executionstore/context_checkpoints_store.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ func (s *Store) PublishContextCheckpoint(
3434
"PublishContextCheckpoint",
3535
func(ctx context.Context) (ContextCheckpointRecord,
3636
error) {
37-
unit, h, err := beginExecution(ctx, s, input.ProjectID, input.AgentID)
37+
unit, h, err := beginExecution(ctx, s, input.ProjectID, input.AgentID, false)
3838
if err != nil {
3939
return ContextCheckpointRecord{}, err
4040
}

‎internal/storage/executionstore/context_checkpoints_terminal_failure.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,7 @@ func (s *Store) RecordTerminalCompactionFailure(
3636
_, err := agentexecution.Retry(ctx,
3737
"RecordTerminalCompactionFailure",
3838
func(ctx context.Context) (struct{}, error) {
39-
unit, h, err := beginExecution(ctx, s, input.ProjectID, input.AgentID)
39+
unit, h, err := beginExecution(ctx, s, input.ProjectID, input.AgentID, true)
4040
if err != nil {
4141
return struct{}{}, err
4242
}

‎internal/storage/executionstore/execution_mutation.go‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,12 +12,18 @@ func beginExecution(
1212
ctx context.Context,
1313
s *Store,
1414
projectID, agentID uuid.UUID,
15+
includeParent bool,
1516
) (*agentexecution.Unit, *agentexecution.Handle, error) {
1617
unit, err := s.cell.Begin(ctx)
1718
if err != nil {
1819
return nil, nil, err
1920
}
20-
plan, err := unit.PlanAgents(ctx, []lifecyclelock.AgentRef{{ProjectID: projectID, AgentID: agentID}}, nil)
21+
var plan agentexecution.MutationPlan
22+
if includeParent {
23+
plan, err = unit.PlanAgentFamily(ctx, projectID, agentID, nil)
24+
} else {
25+
plan, err = unit.PlanAgents(ctx, []lifecyclelock.AgentRef{{ProjectID: projectID, AgentID: agentID}}, nil)
26+
}
2127
if err == nil {
2228
err = unit.LockAgents(ctx, plan)
2329
}

‎internal/storage/executionstore/execution_unit_lock_order_integration_test.go‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -78,7 +78,7 @@ func assertExecutionAgentOrder(
7878
)
7979
}
8080

81-
func TestModelMutationsFenceWithoutParentLock(t *testing.T) {
81+
func TestModelMutationsPlanLocksBeforeFencing(t *testing.T) {
8282
for _, parentFirst := range []bool{true, false} {
8383
t.Run(fmt.Sprintf("parent_first_%t", parentFirst), func(t *testing.T) {
8484
f, parent, child := executionLockOrderAgents(t, parentFirst)
@@ -88,7 +88,11 @@ func TestModelMutationsFenceWithoutParentLock(t *testing.T) {
8888
"begin compaction", "replace compaction", "terminal compaction",
8989
} {
9090
t.Run(operation, func(t *testing.T) {
91-
assertExecutionAgentOrder(t, f, child, parent, func(ctx context.Context) error {
91+
first, second := child, parent
92+
if operation == "terminal compaction" && parentFirst {
93+
first, second = parent, child
94+
}
95+
assertExecutionAgentOrder(t, f, first, second, func(ctx context.Context) error {
9296
s := f.Store.Execution()
9397
switch operation {
9498
case "claim compaction":

‎internal/storage/executionstore/export_integration_test.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -45,7 +45,7 @@ func (s *Store) AcquireAgentRuntimeLock(
4545
return agentexecution.Retry(ctx,
4646
"AcquireAgentRuntimeLock",
4747
func(ctx context.Context) (AgentRuntimeLockRecord, error) {
48-
unit, h, err := beginExecution(ctx, s, projectID, agentID)
48+
unit, h, err := beginExecution(ctx, s, projectID, agentID, false)
4949
if err != nil {
5050
return AgentRuntimeLockRecord{}, err
5151
}
@@ -87,7 +87,7 @@ func (s *Store) RepairExecutionForTest(ctx context.Context, projectID, agentID u
8787
_, retryErr := agentexecution.Retry(ctx,
8888
"RepairExecutionForTest",
8989
func(ctx context.Context) (struct{}, error) {
90-
unit, h, err := beginExecution(ctx, s, projectID, agentID)
90+
unit, h, err := beginExecution(ctx, s, projectID, agentID, false)
9191
if err != nil {
9292
return struct{}{}, err
9393
}
@@ -122,7 +122,7 @@ func (s *Store) NextAgentModelWork(
122122
retried, retryErr := agentexecution.Retry(ctx,
123123
"NextAgentModelWork",
124124
func(ctx context.Context) (transactionResult, error) {
125-
unit, h, err := beginExecution(ctx, s, projectID, agentID)
125+
unit, h, err := beginExecution(ctx, s, projectID, agentID, false)
126126
if err != nil {
127127
return transactionResult{}, err
128128
}

0 commit comments

Comments
 (0)