fix(replan): retain original goals and complete scoped evidence

benchmark/edgebench/README.md

@@ -223,8 +223,13 @@ and never justify blind resubmission. This does not add restart persistence.
 
 New non-game native trials include brief guidance to keep researching the original
 task objective, treating implementation, passing checks and submission as milestones.
-The same text reaches the startup prompt, workspace `AGENTS.md` and LoopX Goal;
-the complete original task remains the planning input. Trial budgets and deadlines
+This guidance reaches the startup prompt and workspace `AGENTS.md`. New LoopX
+Goals now retain the task's original `agent_query` as their objective, so planning
+and replan readback preserve its stated priorities rather than a generic reminder.
+If the query points to task files, those files remain required source material;
+this does not extract or invent a structured acceptance contract. Existing Goals
+and restored checkpoints retain their captured objective without migration.
+Trial budgets and deadlines
 stay with the existing runner, without extra agent-facing budget instructions.
 This changes guidance for new native attempts across all five workers, preserving
 grading, cooldown, selection, blind/best-only controls, game prompts and frozen runs.

benchmark/edgebench/prompts.py

@@ -7,8 +7,8 @@
 from __future__ import annotations
 
 
-def native_task_objective() -> str:
-    """Keep continued research, rather than Todo completion, in the Goal."""
+def native_research_guidance() -> str:
+    """Encourage continued research without replacing the task-owned objective."""
     return (
         "Continue researching and improving the original task objective. "
         "Treat implementation, passing checks, and submission as milestones; "
@@ -82,7 +82,7 @@ def native_task_instructions(submit_paths: list[str], *,
         "or feedback, repeatedly retry a refused submission, or make meaningless "
         "changes to fill the interval. Experimentation is encouraged; local evidence "
         "and evaluator feedback have different coverage.\n\n"
-        f"{native_task_objective()}\n\n"
+        f"{native_research_guidance()}\n\n"
         "### Scoring\n\n"
         f"Native selection policy: `{selection}`; score direction: `{score_direction}`. "
         "The task's grading and final-selection rules determine your result; feedback "

benchmark/edgebench/run.py

@@ -29,7 +29,6 @@
 from benchmark.runtime.source import source_pins
 from benchmark.edgebench.prompts import (
     blind_task_prompt, best_only_task_prompt, native_task_prompt, native_task_instructions,
-    native_task_objective,
 )
 from benchmark.edgebench.online_sampling import OnlineSampler
 from benchmark.edgebench.feedback import BestOnlyFeedback, FEEDBACK_MODES, validate_best_only
@@ -233,7 +232,7 @@ def main(argv=None):
             selection=task.judge.selection, score_direction=task.judge.score_direction)
         feedback_prompt = native_task_prompt(task.work.agent_query, task.submit_paths, **native_policy)
         workspace_instructions = native_task_instructions(task.submit_paths, **native_policy)
-        task_objective = native_task_objective()
+        task_objective = task.work.agent_query
     config = SForgeConfig(
         agent_api_base_url=api_base_url, agent_api_key=api_key,
         agent_model=args.model, agent_effort=args.effort,

benchmark/runtime/sforge.py

@@ -179,9 +179,10 @@ def __init__(self, config, *, profile: str, cwd: str,
             raise ValueError("Workspace feedback instructions must match the native task wrapper")
         if task_objective is not None and (
             not isinstance(task_objective, str) or not task_objective.strip()
-            or workspace_instructions is None or task_objective not in workspace_instructions
+            or workspace_instructions is None or feedback_prompt is None
+            or task_objective not in feedback_prompt
         ):
-            raise ValueError("Task objective must occur in the native workspace instructions")
+            raise ValueError("Task objective must occur in the native task prompt")
         self.feedback = feedback
         self.feedback_prompt = feedback_prompt
         self.workspace_instructions = workspace_instructions

benchmark/tests/test_edgebench_prompts.py

@@ -10,7 +10,7 @@
 
 from benchmark.edgebench.prompts import (
     blind_task_prompt, best_only_task_prompt, native_task_prompt, native_task_instructions,
-    native_task_objective,
+    native_research_guidance,
 )
 
 
@@ -81,7 +81,7 @@ def test_each_worker_receives_and_records_selected_prompt(tmp_path, monkeypatch,
         instructions = native_task_instructions(["solver.py"],
             submission_cooldown=3600, eval_interval=0, internet=False,
             selection="pass_rate_first", score_direction="maximize")
-        objective = native_task_objective()
+        objective = "Complete the supplied task."
     else:
         render = best_only_task_prompt if feedback == "best-only" else blind_task_prompt
         prompt = render("Complete the supplied task.", ["solver.py"])
@@ -128,7 +128,7 @@ async def configure(*args, **kwargs):
         assert agents.read_text().startswith("Task-owned instructions: preserve the interface.\n")
         assert agents.read_text().endswith(instructions)
         assert prompt.startswith(instructions)
-        assert objective == worker.task_objective and objective in agents.read_text()
+        assert objective == worker.task_objective and objective in prompt
         assert "`sforge-submit --result ID --details`" in agents.read_text()
         assert "`sforge-submit --help`" in prompt
         assert (tmp_path / "workspace-AGENTS.md").read_text() == agents.read_text()
@@ -151,13 +151,13 @@ def test_native_research_guidance_preserves_original_task_without_budget_injecti
                    selection="score_first", score_direction="maximize")
     policy = native_task_instructions(["solution.py"], **options)
     assert native_task_prompt(query, ["solution.py"], **options) == policy + "---\n\n" + query + "\n"
-    assert native_task_objective() in policy
+    assert native_research_guidance() in policy
     assert "total trial budget" not in policy
     assert "remaining time" not in policy
 
 
 @pytest.mark.parametrize("objective", ["", "  ", 64800, "Unrelated objective"])
-def test_native_goal_transport_rejects_objective_outside_task_policy(objective, tmp_path, monkeypatch):
+def test_native_goal_transport_rejects_objective_outside_task_prompt(objective, tmp_path, monkeypatch):
     pytest.importorskip("sforge")
     pytest.importorskip("harbor")
     from sforge.harness.config import SForgeConfig
@@ -191,7 +191,8 @@ def test_native_research_objective_and_full_planning_input_survive_real_goal_boo
     cli.chmod(0o755)
     for name, value in {"_CONTROL": control, "_TASK_DOC": control / "task.md",
                         "_REGISTRY": control / "registry.json", "_CLI": cli,
-                        "_LOOPX_RUNTIME": tmp_path / "runtime"}.items():
+                        "_LOOPX_RUNTIME": tmp_path / "runtime",
+                        "_SRC": Path(__file__).resolve().parents[2]}.items():
         monkeypatch.setattr(harbor, name, str(value))
     # Exercise real shell, task uploads, bootstrap/configure and planner without
     # staging binaries, contacting a model or creating a task/grader container.
@@ -214,12 +215,13 @@ async def upload_file(self, source, target):
     monkeypatch.setenv("CODEX_AUTH_JSON_PATH", "/synthetic-credential")
     options = dict(submission_cooldown=3600, eval_interval=0, internet=False,
                    selection="score_first", score_direction="maximize")
-    query = "Optimize original metric A while preserving hard constraint B."
+    query = ("Optimize metric A first, then metric C, while preserving hard constraint B.\n"
+             "Do not replace the original priorities with a local proxy.")
     policy = native_task_instructions(["solution.py"], **options)
     prompt = native_task_prompt(query, ["solution.py"], **options)
     worker = SForgeWorker(SForgeConfig(agent_model="fixture", agent_effort="xhigh"),
         profile=profile, cwd=str(project), feedback_prompt=prompt, workspace_instructions=policy,
-        task_objective=native_task_objective())
+        task_objective=query)
     worker.install_stop_hook(None, None, tmp_path, None)
     worker.environment = Environment()
     remote = control / "entry.md"
@@ -229,7 +231,9 @@ async def upload_file(self, source, target):
         state = Path(registry["goals"][0]["state_file"])
         if not state.is_absolute():
             state = project / state
-        assert worker.task_objective in state.read_text()
+        from loopx.control_plane.goals.active_state_metadata import split_state_frontmatter
+        assert split_state_frontmatter(state.read_text())[0]["objective"] == query
+        assert native_research_guidance() not in state.read_text()
         assert remote.read_text() == prompt
         assert policy in (project / "AGENTS.md").read_text()
         packet = task_plan_packet(env | worker.runtime._worker_env(cwd=str(project)),
@@ -238,8 +242,22 @@ async def upload_file(self, source, target):
         assert packet["text"].endswith(prompt)
         assert worker.task_objective in packet["text"] and query in packet["text"]
         assert packet["existing_todos"] == [] and packet["execution_handoff"]["spends_quota"] is False
+        # Real typed replan owner reads the bootstrapped registry/state, not a
+        # hand-built objective fixture. It must retain both ordered priorities
+        # and the trailing constraint through the scoped source projection.
+        readback = subprocess.run([sys.executable, "-c",
+            "import json,sys; from loopx.control_plane.work_items.replan_context_codec "
+            "import project_replan_context; print(json.dumps(project_replan_context("
+            "goal_id='benchmark-goal',agent_id='benchmark-agent',runs=(),"
+            "source_status={'registry':sys.argv[1],'runtime_root':sys.argv[2],'run_history':{'runs':[]}})))",
+            harbor._REGISTRY, harbor._LOOPX_RUNTIME], env=env, cwd=project,
+            capture_output=True, text=True, timeout=90)
+        assert readback.returncode == 0, readback.stderr
+        core = json.loads(readback.stdout)["core_goal"]
+        assert core["objective"] == " ".join(query.split())
+        assert core["objective_source"] == "active_state" and core["acceptance_contract"] is None
         assert worker.format_run_cmd(str(remote), internet=False, resume=True) == command
-        assert (project / "AGENTS.md").read_text().count(native_task_objective()) == 1
+        assert (project / "AGENTS.md").read_text().count(native_research_guidance()) == 1
     finally:
         subprocess.run([sys.executable, "-c", "from loopx.control_plane.effect_runtime import "
             "effect_runtime_result; effect_runtime_result('runtime.shutdown',{},retry_safe=False)"],

benchmark/tests/test_sforge_runtime.py

@@ -667,7 +667,7 @@ def test_invalid_checkpoint_declaration_rejects_before_trial_registration(tmp_pa
     ("lean_analysis_proofs", [], 43200, 0),
     ("portfolio_risk_calibration", ["--timeout", "1800", "--eval-interval", "60"], 1800, 60),
     ("lean_analysis_proofs", ["--eval-interval", "0"], 43200, 0)])
-def test_edgebench_receipt_records_resolved_entry_and_enabled_treatment(tmp_path, monkeypatch, enabled, task, timeout_args, expected, interval, entry, cadence_args, field, count):
+def test_edgebench_native_goal_preserves_task_with_resolved_entry_and_treatment(tmp_path, monkeypatch, enabled, task, timeout_args, expected, interval, entry, cadence_args, field, count):
     pytest.importorskip("sforge")
     pytest.importorskip("harbor")
     from types import SimpleNamespace
@@ -694,7 +694,10 @@ def stop_before_solver(**kwargs):
         assert kwargs["eval_interval"] == interval
         agent = kwargs["agent"]
         assert "total trial budget" not in agent.workspace_instructions
-        assert agent.task_objective in agent.workspace_instructions
+        assert agent.task_objective == "fixture task"
+        assert agent.task_objective in agent.feedback_prompt
+        assert agent.task_objective not in agent.workspace_instructions
+        assert "Continue researching" in agent.workspace_instructions
         assert agent.feedback_prompt.startswith(agent.workspace_instructions)
         raise RuntimeError("synthetic launch failure")
     monkeypatch.setattr(run, "run_agent", stop_before_solver)

docs/reference/protocols/agent-scoped-evidence-ledger-v0.md

@@ -66,7 +66,24 @@ revision and status remain a separate `acceptance_contract`: a selected-work
 contract and an Agent's current task do not redefine the whole Goal. This read
 model creates no new Goal or acceptance authority.
 
-Evidence retains public-safe validation/action summaries linked to typed
+The current Goal and selected evidence are scoped decision input. Replan keeps
+complete objective, validation and action text, including task/material paths
+and credential-like strings present in the source; it does not run the public
+publication sanitizer or cut those fields to a 240-character display prefix.
+This changes formerly suppressed or shortened replan fields for both ordinary
+and stride-enabled replan. The existing source scoping, bounded observation
+selection, omitted-source reads and private snapshot transport remain intact.
+These packets are not public artifacts: export/upload/publication boundaries
+still require their own redaction and validation.
+
+Each admitted delivery Turn also receives the existing full Goal-state read in
+`interaction_contract.agent_channel.required_reads`, with configured scoped
+acceptance read separately. `planning_horizon` is an optional bounded view of
+work, relations and acceptance gaps; its absence does not remove Goal
+persistence or replace those reads. Reading the Goal does not grant execution
+authority or prove that a subsequent decision used it.
+
+Evidence retains scoped validation/action text linked to typed
 `coverage_ledger` observations by fingerprint. Other Agents and unattributed
 records cannot supply proof for a scoped Agent. Counts describe the available
 source, not complete Goal acceptance. Omitted observations expose counts, time
@@ -87,7 +104,7 @@ loopx --format json history --goal-id <goal-id> --agent-id <agent-id> --evidence
 ```
 
 Execute the supplied action only when more detail is needed. It resolves one
-public-safe compact record in the same Goal and Agent; a changed or unavailable reference fails explicitly and asks the caller to refresh context.
+scoped compact-index record in the same Goal and Agent; a changed or unavailable reference fails explicitly and asks the caller to refresh context.
 It never falls back to another Agent or the latest unrelated record. It reads no
 raw transcript and creates no read receipt. Ordinary `history` remains the
 operator's compact run-history view; `--agent-id` scopes that view as well.

loopx/control_plane/work_items/replan_context.ts

@@ -81,13 +81,13 @@ function repeatedAccounting(rows: JsonObject[]): Map<JsonObject, JsonObject> {
   return repeated;
 }
 
-function publicEvidenceRow(row: JsonObject): JsonObject {
-  const {observed_at: _time, _source_facts: _source, ...publicRow} = row;
-  return publicRow;
+function readableEvidenceRow(row: JsonObject): JsonObject {
+  const {observed_at: _time, _source_facts: _source, ...readableRow} = row;
+  return readableRow;
 }
 
 function evidenceFactsKey(row: JsonObject): string {
-  const {generated_at: _time, evidence_ref: _ref, ...facts} = publicEvidenceRow(row);
+  const {generated_at: _time, evidence_ref: _ref, ...facts} = readableEvidenceRow(row);
   return digest(facts);
 }
 
@@ -156,7 +156,7 @@ function recordedDecision(row: JsonObject): JsonObject | null {
   try { stride = normalizeRecordedStrideDecision(objectOrNull(source.agent_vision)?.stride_decision); }
   catch { return null; }
   const autonomous = objectOrNull(source.settlement_identity)?.schema_version === SCOPED_SETTLEMENT_IDENTITY_SCHEMA_VERSION;
-  const decision = {history_evidence_ref: "replan-evidence-" + digest(publicEvidenceRow(row)),
+  const decision = {history_evidence_ref: "replan-evidence-" + digest(readableEvidenceRow(row)),
     generated_at: row.generated_at, todo_id: autonomous ? null : row.todo_id, turn_instance_id: source.turn_instance_id,
     ...(autonomous ? {replan_obligation_id: source.replan_obligation_id} : {}),
     goal_ref: ref ? {goal_id: ref.goal_id, goal_instance_id: ref.goal_instance_id} : null,
@@ -203,7 +203,7 @@ function decisionFollowthrough(rows: JsonObject[], goalFacts: unknown, fromFullI
       !Number.isSafeInteger(objectOrNull(row._source_facts)?.history_position)).length,
     recorded_progress_count: observations.length,
     observations: observations.slice(0, 6).map(row => ({
-      evidence_ref: "replan-evidence-" + digest(publicEvidenceRow(row)),
+      evidence_ref: "replan-evidence-" + digest(readableEvidenceRow(row)),
       generated_at: row.generated_at, todo_id: row.todo_id,
       coverage_ref: (row.progress_observation as JsonObject).fingerprint,
     })),
@@ -299,12 +299,12 @@ export function projectReplanContext(value: unknown): JsonObject {
     }
   }
   rows.sort((a, b) => Number(b.observed_at) - Number(a.observed_at) ||
-    JSON.stringify(publicEvidenceRow(a)).localeCompare(JSON.stringify(publicEvidenceRow(b))));
+    JSON.stringify(readableEvidenceRow(a)).localeCompare(JSON.stringify(readableEvidenceRow(b))));
   const unique = new Map<string, JsonObject>();
   for (const row of rows) {
-    const publicRow = publicEvidenceRow(row);
-    const ref = "replan-evidence-" + digest(publicRow);
-    if (!unique.has(ref)) unique.set(ref, {...publicRow, evidence_ref: ref});
+    const readableRow = readableEvidenceRow(row);
+    const ref = "replan-evidence-" + digest(readableRow);
+    if (!unique.has(ref)) unique.set(ref, {...readableRow, evidence_ref: ref});
   }
   const evidence = [...unique.values()];
   if (request.operation === "resolve") {

loopx/control_plane/work_items/replan_context_codec.py

@@ -1,4 +1,4 @@
-"""Public-safe history codec for the TypeScript-owned replan evidence view."""
+"""Scoped source codec for the TypeScript-owned replan evidence view."""
 from __future__ import annotations
 
 from collections.abc import Iterable, Mapping
@@ -8,11 +8,16 @@
 from typing import Any
 
 from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result
-from ..runtime.public_safety import public_safe_compact_text
 from ..runtime.time import parse_timestamp
 from ..runtime.agent_evidence_history import goal_history_runs
 
 
+def _source_text(value: Any) -> str | None:
+    """Keep decision input complete; publication owns its own redaction."""
+    text = str(value or "").strip()
+    return text or None
+
+
 def replan_history_from_status(status: Mapping[str, Any], goal_id: str) -> list[dict[str, Any]]:
     history = status.get("run_history")
     if not isinstance(history, Mapping):
@@ -73,7 +78,7 @@ def replan_evidence_rows(runs: Iterable[Mapping[str, Any]], *, goal_id: str, age
             "progress_observation": progress,
         }
         for key in ("classification", "delivery_outcome", "health_check", "recommended_action", "todo_id"):
-            text = public_safe_compact_text(run.get(key), limit=240)
+            text = _source_text(run.get(key))
             if text:
                 row[key] = text
         if "settlement_identity" in run or "quota_spend_commit" in run or (include_decisions and "agent_vision" in run):
@@ -111,8 +116,7 @@ def _goal_facts(status: Mapping[str, Any] | None, goal_id: str, acceptance: Mapp
     def goal_text(value: Any) -> str | None:
         if value is not None and not isinstance(value, str):
             raise ValueError("Goal objective must be text")
-        # Preserve the complete core objective, including trailing constraints.
-        return public_safe_compact_text(value, limit=max(1, len(value or "")))
+        return _source_text(value)
 
     facts: dict[str, Any] = {}
     if status and status.get("registry"):