[ai] Route follow-ups beside the active task

.agents/skills/continue/SKILL.md

@@ -80,11 +80,11 @@ markers mean discovery routing published new tasks but its separate
 consolidation pass did not finish. Spawn the consolidation worker described
 below with the source task and batch ids recorded in the marker, then refresh
 the queue. A repeated pre-commit race may remain pending for the next
-invocation; record that marker as attempted and do not spin. If a task is
-already active, its expected local phase files make the shared AI slot unsafe
-for consolidation: freeze and finish that active task first, then recover all
-pending markers at its clean canonical `Approve`, `Block`, or completed split
-routing boundary before selecting more work.
+invocation; record that marker as attempted and do not spin. Consolidation publishes
+through the checkout's routing worktree (`route-ensure`,
+`consolidate-publish --routing`), so an active task's slot phase files do not
+defer recovery: recover pending markers whenever no other routing or
+consolidation transaction is in flight.
 
 After recovering pending consolidations and before freezing a new batch, route
 this checkout's `own_split_required` task, when any, through the dedicated split
@@ -450,9 +450,13 @@ artifact-based assessment.
 
 ## Route discovered follow-ups
 
-After every canonical `Approve` or `Block`, read `work/result.md`. Route before
-selecting more shared work whenever it lacks `work/discovered-routed.md` and
-either says `Discovered: present` or carries a non-`none` `Unverified:` value.
+After every canonical `Approve` or `Block`, read `work/result.md`. Route
+whenever it lacks `work/discovered-routed.md` and either says
+`Discovered: present` or carries a non-`none` `Unverified:` value. Routing runs
+beside task selection, not before it: spawn the worker, then continue
+selecting and starting recorded batch work immediately. Keep at most one
+routing worker in flight, and never reach the invocation's normal stop while
+any routing or consolidation is unlanded.
 Both are unfinished work leaving the pipeline; the only difference is that
 `Discovered:` names work nobody has started and `Unverified:` names behavior that
 already shipped without proof. An approved task whose unverified behavior was
@@ -466,6 +470,16 @@ in `.agents/skills/process-inbox/SKILL.md`, but not to call `prepare`,
 must not edit Telegram source, start tasks, or implement work. Give it the
 current ordered `batch_task_ids` for the pending consolidation marker.
 
+The worker edits and publishes through the checkout's routing worktree, never
+the slot: it starts from `workspace.py route-ensure` (which creates and syncs
+the `routing/<tag>` worktree), writes every artifact there, and publishes with
+`workspace.py route-publish --source-task <id> --path <path> ...`, which
+stages exactly the named paths, commits `Route follow-ups from <id>`, and
+rebases onto canonical master before pushing. The slot worktree stays
+untouched, so an active performer's local phase state never conflicts. After a
+crash, an unpublished routing commit is resumed by rerunning
+`route-publish --source-task <id>` with no paths.
+
 The worker must deduplicate existing tasks, create independently testable
 unclaimed `todo` tasks and justified project updates, write a discovery
 receipt, and write the source task's routing marker. When it creates at least
@@ -578,7 +592,8 @@ coverage task already measured a deviation, route only the repair with the
 measured expected and actual values; do not create another measurement of the
 same gap.
 
-After validating the discovery receipt, append only the task ids created from
+After the routing lands on canonical master and its discovery receipt
+validates, append only the task ids created from
 that result to `discovered_task_ids` and `batch_task_ids`, preserving routing
 order. This is the only way the frozen batch grows. Apply the same rule
 transitively when a discovered task later reports its own follow-ups.
@@ -588,7 +603,9 @@ queue refreshes do not join the batch.
 ## Consolidate pending tasks after discovery
 
 Whenever discovery routing publishes `work/consolidation-pending.md`, run one
-fresh consolidation pass before selecting the next task. Do not run it for a
+fresh consolidation pass once that routing has landed; it runs beside the next
+task, publishing through the routing worktree with
+`consolidate-publish --routing`, with at most one consolidation in flight. Do not run it for a
 receipt-only routing or a routing that only reused existing tasks. Do not reuse
 the performer or routing worker: spawn one disposable worker with
 `fork_turns: "none"`, instruct it not to delegate, and give it `source_root`,
@@ -612,11 +629,9 @@ this invocation, and continue without treating the optimization as a blocker.
 If the worker created a commit that cannot be published safely, preserve it and
 hard-stop exactly as for discovery routing.
 
-At every clean canonical `Approve`, `Block`, or completed split routing
-boundary, process any older
-pending marker deferred by an active startup task before selecting more work,
-then process the marker just created by that task's routing. Attempt each marker
-at most once per invocation.
+Process any older pending marker whenever no consolidation is in flight, then
+the marker created by the newest routing, in publication order. Attempt each
+marker at most once per invocation.
 
 For each published old-to-new mapping, rewrite `batch_task_ids` by placing the
 replacement at the earliest position occupied by any of its sources and removing

.agents/skills/process-inbox/scripts/workspace.py

@@ -285,6 +285,55 @@ def sync_branch_worktree(config, worktree_key, branch_key, label):
 		run_git(worktree, "merge", "--ff-only", "master")
 
 
+def routing_worktree_config(args, create=False):
+	config = repository_config(args, create=create)
+	tag = config["checkout_tag"]
+	branch = f"routing/{tag}"
+	worktree = Path(config["worktrees_root"]) / f"{tag}-routing"
+	linked_worktree(config, worktree, branch, create, "routing AI worktree")
+	return {
+		**config,
+		"routing_worktree": str(worktree),
+		"routing_branch": branch,
+	}
+
+
+def sync_routing_worktree(config):
+	sync_branch_worktree(
+		config,
+		"routing_worktree",
+		"routing_branch",
+		"ai-tdesktop routing worktree",
+	)
+
+
+def sync_routing_canonical(config):
+	update_main_from_origin(config)
+	sync_routing_worktree(config)
+
+
+def routing_unpublished_count(config):
+	worktree = Path(config["routing_worktree"])
+	counts = run_git(
+		worktree,
+		"rev-list",
+		"--left-right",
+		"--count",
+		f"master...{config['routing_branch']}",
+	).stdout.strip().split()
+	return int(counts[1])
+
+
+def publish_routing(config, validate=None):
+	return publish_worktree(
+		config,
+		"routing_worktree",
+		"routing_branch",
+		"ai-tdesktop routing worktree",
+		validate,
+	)
+
+
 def has_origin(path):
 	return run_git(path, "remote", "get-url", "origin", check=False).returncode == 0
 
@@ -3907,6 +3956,10 @@ def command_task_content_digest(args):
 
 def command_consolidate_publish(args):
 	config = worktree_config(args, create=True)
+	if getattr(args, "routing", False):
+		routing = routing_worktree_config(args, create=True)
+		config["slot_worktree"] = routing["routing_worktree"]
+		config["slot_branch"] = routing["routing_branch"]
 	slot = Path(config["slot_worktree"])
 	if not TASK_ID_PATTERN.fullmatch(args.source_task):
 		raise WorkspaceError(f"Invalid source task: {args.source_task!r}")
@@ -4324,6 +4377,90 @@ def command_ensure(args):
 	print(json.dumps(config, indent=2, sort_keys=True))
 
 
+def command_route_ensure(args):
+	config = routing_worktree_config(args, create=True)
+	worktree = Path(config["routing_worktree"])
+	dirty = changed_paths(worktree)
+	unpublished = routing_unpublished_count(config)
+	if not dirty and not unpublished:
+		sync_routing_canonical(config)
+	print(json.dumps({
+		**config,
+		"routing_dirty": dirty,
+		"routing_unpublished": unpublished,
+	}, indent=2, sort_keys=True))
+
+
+def command_route_publish(args):
+	config = routing_worktree_config(args, create=True)
+	worktree = Path(config["routing_worktree"])
+	if not TASK_ID_PATTERN.fullmatch(args.source_task):
+		raise WorkspaceError(f"Invalid source task: {args.source_task!r}")
+	changes = changed_paths(worktree)
+	if not changes:
+		if not routing_unpublished_count(config):
+			raise WorkspaceError("The routing worktree has nothing to publish")
+		published = publish_routing(config)
+		print(json.dumps({
+			"committed": False,
+			"published": bool(published),
+			"routing_branch": config["routing_branch"],
+		}, indent=2, sort_keys=True))
+		return
+	if not args.paths:
+		raise WorkspaceError("Publication paths are required when changes exist")
+	paths = sorted({normalized_publish_path(value) for value in args.paths})
+	marker = f"tasks/{args.source_task}/work/discovered-routed.md"
+	if not path_is_covered(marker, paths):
+		raise WorkspaceError(
+			"The discovered-routed marker is not covered by a publication path"
+		)
+	unexpected = [path for path in changes if not path_is_covered(path, paths)]
+	if unexpected:
+		raise WorkspaceError(
+			"Routing worktree changes are outside the explicit publication paths: "
+			+ ", ".join(unexpected)
+		)
+	for path in paths:
+		if (worktree / path).exists() or any(
+			path_is_covered(change, [path]) for change in changes
+		):
+			run_git(worktree, "add", "-A", "--", path)
+	unstaged = literal_paths(worktree, "diff", "--name-only")
+	untracked = literal_paths(
+		worktree,
+		"ls-files",
+		"--others",
+		"--exclude-standard",
+	)
+	if unstaged or untracked:
+		raise WorkspaceError(
+			"Routing worktree changes remain unstaged: "
+			+ ", ".join(sorted(set(unstaged + untracked)))
+		)
+	committed = run_git(
+		worktree,
+		"diff",
+		"--cached",
+		"--quiet",
+		check=False,
+	).returncode != 0
+	if committed:
+		run_git(
+			worktree,
+			"commit",
+			"-m",
+			f"Route follow-ups from {args.source_task}",
+		)
+	published = publish_routing(config)
+	print(json.dumps({
+		"committed": committed,
+		"published": bool(published),
+		"routing_branch": config["routing_branch"],
+		"routing_worktree": config["routing_worktree"],
+	}, indent=2, sort_keys=True))
+
+
 def command_inbox_ensure(args):
 	config = inbox_worktree_config(args, create=True)
 	print(json.dumps(config, indent=2, sort_keys=True))
@@ -4650,8 +4787,23 @@ def parse_args():
 	)
 	inbox_publish.set_defaults(handler=command_inbox_publish)
 
+	route_ensure = subparsers.add_parser("route-ensure")
+	add_common_arguments(route_ensure)
+	route_ensure.set_defaults(handler=command_route_ensure)
+
+	route_publish = subparsers.add_parser("route-publish")
+	add_common_arguments(route_publish)
+	route_publish.add_argument("--source-task", required=True)
+	route_publish.add_argument(
+		"--path",
+		action="append",
+		dest="paths",
+	)
+	route_publish.set_defaults(handler=command_route_publish)
+
 	consolidate_publish = subparsers.add_parser("consolidate-publish")
 	add_common_arguments(consolidate_publish)
+	consolidate_publish.add_argument("--routing", action="store_true")
 	consolidate_publish.add_argument("--source-task", required=True)
 	consolidate_publish.add_argument("--receipt")
 	consolidate_publish.add_argument(