Job: target_node claim so pawns don't cluster on the same target

Visible bug: with 1 wall ghost queued, all 3 pawns picked the same site;
2 stood idle while 1 built. Same shape would affect chop/mine/haul/etc.

Design: Job carries an untyped target_node ref (the tree/rock/build-site/
crop/item/patient/workbench/etc the job is acting on). Job.is_target_taken_
by_other(target, excluding_pawn) does an O(pawns) scan of live job state to
ask 'is anyone else already working this?'. Each WorkProvider's find_best_
for() now skips claimed targets in its scan and sets j.target_node before
returning. No per-entity claim state, no .claim()/.release() bookkeeping,
no save-format change — target_node is intentionally not serialized
because pawns re-decide and re-bind naturally after load.

Providers updated: construction / chop / mine / plant (harvest path) /
hauling (item AND corpse) / cleaning (target is Vector2i tile not Node;
field is untyped, doc'd) / doctor / crafting (workbench).

Not touched: rest (everyone shares the rest tile, that's fine), eat /
sleep (food and beds have their own availability gates; flagged as a
followup if multi-pawn food contention surfaces).

Verified MCP runtime: fresh boot, 3 pawns picked 3 distinct wall sites
(44,28)/(45,28)/(44,27) with distinct target_node refs.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

scenes/ai/chop_provider.gd

@@ -34,6 +34,8 @@ func find_best_for(pawn) -> Job:
 	for tree in World.trees:
 		if not tree.is_choppable():
 			continue
+		if Job.is_target_taken_by_other(tree, pawn):
+			continue
 		var d: int = abs(tree.tile.x - pawn.tile.x) + abs(tree.tile.y - pawn.tile.y)
 		if d < best_dist:
 			best_dist = d
@@ -44,6 +46,7 @@ func find_best_for(pawn) -> Job:
 
 	var j := Job.new()
 	j.label = "Chop tree at %s" % best.tile
+	j.target_node = best
 	j.toils.append(Toil.walk_to(best.tile))
 	j.toils.append(Toil.interact(best.get_path(), &"on_chop_tick"))
 	return j

scenes/ai/cleaning_provider.gd

@@ -50,6 +50,11 @@ func find_best_for(pawn) -> Job:
 		var dirt_val: float = float(ds.dirt_map[tile])
 		if dirt_val < DIRTY_THRESHOLD:
 			continue
+		# target_node stores the Vector2i tile coordinate (field is untyped — accepts
+		# non-Node values). Dirty tiles have no scene Node; the tile position itself
+		# is the unique claim key.
+		if Job.is_target_taken_by_other(tile, pawn):
+			continue
 		var d: int = abs(tile.x - pawn.tile.x) + abs(tile.y - pawn.tile.y)
 		if d < best_dist:
 			best_dist = d
@@ -60,6 +65,7 @@ func find_best_for(pawn) -> Job:
 
 	var j := Job.new()
 	j.label = "Clean (%d,%d)" % [best_tile.x, best_tile.y]
+	j.target_node = best_tile
 	j.toils.append(Toil.walk_to(best_tile))
 	j.toils.append(Toil.clean_at(best_tile))
 	return j

scenes/ai/construction_provider.gd

@@ -46,6 +46,8 @@ func find_best_for(pawn) -> Job:
 	for site in World.build_queue:
 		if not site.is_buildable():
 			continue
+		if Job.is_target_taken_by_other(site, pawn):
+			continue
 		var d: int = abs(site.tile.x - pawn.tile.x) + abs(site.tile.y - pawn.tile.y)
 		if d < best_dist:
 			best_dist = d
@@ -66,6 +68,7 @@ func find_best_for(pawn) -> Job:
 
 	var j := Job.new()
 	j.label = "Build %s at %s" % [best.label(), best.tile]
+	j.target_node = best
 	j.toils.append(Toil.walk_to(stand_tile))
 	j.toils.append(Toil.build_at(best.get_path()))
 	return j

scenes/ai/crafting_provider.gd

@@ -55,6 +55,9 @@ func find_best_for(pawn) -> Job:
 		# Duck-type guard: skip workbenches that aren't fully set up yet.
 		if not wb.get("_completed"):
 			continue
+		# Skip workbenches another pawn has already claimed for crafting.
+		if Job.is_target_taken_by_other(wb, pawn):
+			continue
 
 		for i in wb.bills.size():
 			var b = wb.bills[i]
@@ -91,6 +94,7 @@ func find_best_for(pawn) -> Job:
 
 	var j := Job.new()
 	j.label = "Craft %s at %s" % [best_bill.recipe.label, best_wb.get("label_text") if best_wb.get("label_text") != null else "workbench"]
+	j.target_node = best_wb
 	j.toils.append(Toil.walk_to(src_item.tile))
 	j.toils.append(Toil.pickup())
 	j.toils.append(Toil.walk_to(best_wb.tile))

scenes/ai/doctor_provider.gd

@@ -54,7 +54,8 @@ func find_best_for(pawn) -> Job:
 	if pawn.carried_item != null:
 		return null
 
-	# Find the nearest downed pawn that isn't this pawn.
+	# Find the nearest downed pawn that isn't this pawn and has no other doctor
+	# already assigned (prevents two doctors converging on one wounded pawn).
 	var best_patient = null
 	var best_dist: int = 999999
 	for p in World.pawns:
@@ -62,6 +63,8 @@ func find_best_for(pawn) -> Job:
 			continue
 		if p == pawn:
 			continue
+		if Job.is_target_taken_by_other(p, pawn):
+			continue
 		var d: int = abs(p.tile.x - pawn.tile.x) + abs(p.tile.y - pawn.tile.y)
 		if d < best_dist:
 			best_dist = d
@@ -81,6 +84,7 @@ func find_best_for(pawn) -> Job:
 
 	var j := Job.new()
 	j.label = "Rescue %s → bed at %s" % [best_patient.pawn_name, medical_bed.tile]
+	j.target_node = best_patient
 	# 4-toil sequence per the doctor rescue model above.
 	j.toils.append(Toil.walk_to(best_patient.tile))
 	j.toils.append(Toil.rescue(best_patient.get_path()))

scenes/ai/hauling_provider.gd

@@ -62,6 +62,9 @@ func find_best_for(pawn) -> Job:
 		# Skip items another pawn is already carrying.
 		if item.being_carried:
 			continue
+		# Skip items another pawn has already claimed (target_node = item node).
+		if Job.is_target_taken_by_other(item, pawn):
+			continue
 
 		# Find the best destination for this item type + priority.
 		var dest = _find_best_destination_for(item)
@@ -105,6 +108,9 @@ func find_best_for(pawn) -> Job:
 		# Skip corpses another pawn is already carrying.
 		if corpse.get_meta("being_carried_corpse", false):
 			continue
+		# Skip corpses another pawn has already claimed.
+		if Job.is_target_taken_by_other(corpse, pawn):
+			continue
 
 		var dest = _find_best_destination_for(corpse)
 		if dest == null:
@@ -126,6 +132,9 @@ func find_best_for(pawn) -> Job:
 		return null
 
 	var j := Job.new()
+	# target_node is the item/corpse being picked up — the carry-from side.
+	# Other pawns' HaulingProviders read this to avoid double-claiming the same item.
+	j.target_node = best_item
 	if best_is_corpse:
 		j.label = "Haul corpse '%s' -> (%d,%d)" % [
 			best_item.deceased_name,

scenes/ai/job.gd

@@ -14,6 +14,40 @@ var label: String = ""
 var toils: Array[Toil] = []
 var current_toil_index: int = 0
 
+## The world-space entity this job is acting on (tree, rock, build-site,
+## crop, etc.). Set by the WorkProvider that built the job. Untyped to
+## avoid class_name cycles. Read by sibling providers via
+## Job.is_target_taken_by_other() to prevent multiple pawns claiming
+## the same target.
+##
+## NOT serialized: target_node holds a live Node ref (or Vector2i for tile
+## targets). After load, pawns re-decide and re-bind claims naturally based
+## on restored JobRunner state.
+var target_node = null
+
+
+# ── claim helpers ────────────────────────────────────────────────────────────
+
+## Returns true if `target` is the active target_node of any pawn's current
+## job, excluding `excluding_pawn` (the pawn that's about to claim it).
+## O(pawns) scan — single-threaded sim, no race within a tick because
+## pawns iterate sequentially (earlier pawn commits its job before the
+## next pawn's decision runs).
+##
+## Accepts any value for target (Node, Vector2i, etc.) — uses == comparison.
+static func is_target_taken_by_other(target, excluding_pawn) -> bool:
+	if target == null:
+		return false
+	for p in World.pawns:
+		if p == excluding_pawn:
+			continue
+		if p.job_runner == null or not p.job_runner.has_job():
+			continue
+		if p.job_runner.job.target_node == target:
+			Audit.log("decision", "claim skip: %s already targeting %s" % [p.pawn_name, str(target)])
+			return true
+	return false
+
 
 # ── queries ──────────────────────────────────────────────────────────────────
 

scenes/ai/mine_provider.gd

@@ -35,6 +35,8 @@ func find_best_for(pawn) -> Job:
 	for rock in World.rocks:
 		if not rock.is_mineable():
 			continue
+		if Job.is_target_taken_by_other(rock, pawn):
+			continue
 		var d: int = abs(rock.tile.x - pawn.tile.x) + abs(rock.tile.y - pawn.tile.y)
 		if d < best_dist:
 			best_dist = d
@@ -45,6 +47,7 @@ func find_best_for(pawn) -> Job:
 
 	var j := Job.new()
 	j.label = "Mine rock at %s" % best.tile
+	j.target_node = best
 	j.toils.append(Toil.walk_to(best.tile))
 	j.toils.append(Toil.interact(best.get_path(), &"on_mine_tick"))
 	return j

scenes/ai/plant_provider.gd

@@ -45,6 +45,8 @@ func find_best_for(pawn) -> Job:
 	for crop in World.crops:
 		if not crop.is_harvestable():
 			continue
+		if Job.is_target_taken_by_other(crop, pawn):
+			continue
 		var d: int = abs(crop.tile.x - pawn.tile.x) + abs(crop.tile.y - pawn.tile.y)
 		if d < best_dist:
 			best_dist = d
@@ -55,6 +57,7 @@ func find_best_for(pawn) -> Job:
 
 	var j := Job.new()
 	j.label = "Harvest %s at %s" % [best.crop_kind, best.tile]
+	j.target_node = best
 	j.toils.append(Toil.walk_to(best.tile))
 	j.toils.append(Toil.interact(best.get_path(), &"on_harvest_tick"))
 	return j