Skip to content

Deliver actionable analysis and context feedback after restart #297

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
mikamikasuki wants to merge 5 commits into from
Closed

Deliver actionable analysis and context feedback after restart #297

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
3b4c9c7
Report exact registered selectors in Agent binding feedback
mikamikasuki Oct 11, 2026
3a49a81
Deliver current analysis guards and invalid context query feedback
mikamikasuki Oct 11, 2026
64e2156
Deliver actionable prerequisite replanning feedback
mikamikasuki Oct 12, 2026
e66581f
Merge commit 'c8bf953' into HEAD
mikamikasuki Oct 12, 2026
2116644
Merge commit '1ffb801' into HEAD
mikamikasuki Oct 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 11 additions & 7 deletions research/experiments/study_memory.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,10 @@
from research.agents.context_store import ContextStore


class ContextQueryError(ValueError):
"""A caller supplied an invalid query for existing public evidence."""


class ContextPageCapacityError(ValueError):
"""The selected working request cannot deliver even one source character."""

Expand Down Expand Up @@ -68,30 +72,30 @@ def read_json(self, segment_id, *, pointer='', offset=0, limit=6000,
metadata = {'origin': 'Episode public memory catalog', 'authority': 'metadata'}
else:
if segment_id not in self.store.index['segments']:
raise ValueError('Unknown segment in this episode public memory')
raise ContextQueryError('Unknown segment in this episode public memory')
metadata = self.store.index['segments'][segment_id]
path = self.store.root / (segment_id + '.txt')
value = json.loads(path.read_text(encoding='utf-8'))
if not isinstance(pointer, str) or pointer and not pointer.startswith('/'):
raise ValueError('A JSON Pointer must be empty or start with /')
raise ContextQueryError('A JSON Pointer must be empty or start with /')
if pointer:
for token in pointer.split('/')[1:]:
token = token.replace('~1', '/').replace('~0', '~')
if isinstance(value, list):
if not token.isdigit() or str(int(token)) != token:
raise ValueError('Array JSON Pointer needs a canonical nonnegative index')
raise ContextQueryError('Array JSON Pointer needs a canonical nonnegative index')
try:
value = value[int(token)]
except IndexError:
raise ValueError('JSON Pointer index is outside the original evidence') from None
raise ContextQueryError('JSON Pointer index is outside the original evidence') from None
elif isinstance(value, dict) and token in value:
value = value[token]
else:
raise ValueError('JSON Pointer is absent from the original evidence')
raise ContextQueryError('JSON Pointer is absent from the original evidence')
if isinstance(offset, bool) or not isinstance(offset, int) or offset < 0:
raise ValueError('Page offset must be a nonnegative integer')
raise ContextQueryError('Page offset must be a nonnegative integer')
if isinstance(limit, bool) or not isinstance(limit, int) or limit < 1:
raise ValueError('Page limit must be a positive integer')
raise ContextQueryError('Page limit must be a positive integer')
if response_capacity is not None and (type(response_capacity) is not int or response_capacity < 1):
raise ValueError('Response interface capacity must be a positive integer')
if measure_response is not None and not callable(measure_response):
Expand Down
97 changes: 85 additions & 12 deletions research/experiments/study_runtime.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -246,6 +246,22 @@ def _validate_action_bindings(payload: Mapping[str, Any], actions: Sequence[Any]
and (action.get("configuration_id") is None or binding.get("configuration_id") == action.get("configuration_id"))
and (action.get("seed") is None or binding.get("seed") == action.get("seed"))]
if len(candidates) != 1:
identified = [binding for binding in bindings if isinstance(binding, Mapping)
and action.get("action_work_id") is not None
and str(binding.get("work_id")) == str(action["action_work_id"])]
if len(identified) == 1:
binding = identified[0]
expected = {"kind": binding.get("action_name", name),
"configuration_id": binding.get("configuration_id"), "seed": binding.get("seed")}
conflicts = {field: {"supplied": name if field == "kind" else action.get(field),
"expected": value} for field, value in expected.items()
if (field == "kind" and name != value)
or (field != "kind" and action.get(field) is not None and action[field] != value)}
if conflicts:
raise ValueError(f"Action {index} names registered action_work_id {_json(action['action_work_id'])} "
f"but its selectors conflict: {_json(conflicts)}. Registered selectors: {_json(expected)}. "
"Copy kind from the binding's action_name. Optional configuration_id and seed must copy "
"the binding values exactly, or be null/omitted when selecting by action_work_id.")
raise ValueError(f"Action {index} must identify exactly one registered binding")
binding_id = str(candidates[0].get("work_id", ""))
if not binding_id:
Expand Down Expand Up @@ -1311,7 +1327,7 @@ def _attach_public_context(self, source_payload: Mapping[str, Any], prepared: Ma
"catalog": catalog, "read_operation": "read_context_segment",
"scientific_progress_eligible": False}
if payload.get("messages"):
messages = copy.deepcopy(payload["messages"])
messages = copy.deepcopy(base_messages)
catalog_note = ("Arm-public context catalog (metadata only): " + _json(catalog)
+ "\nUse read_context_segment with the exact segment ID, JSON pointer, offset, and limit to retrieve exact source content. "
"Paging is observational and does not count as a scientific operation.")
Expand Down Expand Up @@ -1786,7 +1802,7 @@ def reject_decision(reason: str, proposed: Any = None) -> dict[str, Any]:
response = {"node_id": str(node_id), "state": "paused_no_progress" if diagnosed else "completed",
"stdout": base["stdout"] + "\n" + _json({"decision_validation_error": reason,
"no_progress_observation": observation}), "exec_time": base["exec_time"],
"exc_type": "DecisionValidationError", "exc_info": reason, "validation_reason": reason,
"exc_type": "DecisionValidationError", "exc_info": {"args": [reason]}, "validation_reason": reason,
"decision": copy.deepcopy(proposed), "semantic_proposal": signature,
"source_frontier": frontier, "no_progress_observation": observation,
"action_results": [], "action_work_ids": [], "pending_work_ids": [], "verified_metrics": []}
Expand Down Expand Up @@ -2245,8 +2261,8 @@ def execute_model_episode(self, item: Mapping[str, Any], *, client: Any,
delivered_messages.append({"role": "user", "content": "Observed results from completed source-bound operations in this episode:\n" + _json(context)})
if payload.get("response_validation_feedback"):
delivered_messages.append({"role": "user", "content":
"The previous actual response was saved in the episode ledger but rejected before any action was executed. "
"Correct the response using the original registered context. Validation feedback: " +
"The previous actual response and observed action results were saved in the episode ledger. "
"Use the registered context and current feedback to make a new decision. Validation feedback: " +
_json(payload["response_validation_feedback"])})
if delivered_messages:
schema_note = "\n\nReturn exactly one JSON object satisfying this decision contract:\n" + _json(payload["decision_schema"])
Expand Down Expand Up @@ -2664,6 +2680,10 @@ def execute_decision_actions(self, episode: Mapping[str, Any], actions: Sequence
# a child action. Keep this at the aggregate boundary so recovery
# does not mistake it for pending GPU work and replay it forever.
response["request_new_decision"] = True
feedback = [result["response_validation_feedback"] for result in results
if result.get("response_validation_feedback")]
if feedback:
response["response_validation_feedback"] = feedback[0] if len(feedback) == 1 else feedback
if len(results) == 1:
response["result"] = results[0].get("result")
return response
Expand Down Expand Up @@ -2694,7 +2714,7 @@ def execute_decision_action(self, episode: Mapping[str, Any], decision: Mapping[
if not work_id:
raise ValueError("Each action binding needs its predeclared action work_id")
if binding.get("analysis_operation") == "read_context_segment":
from .study_memory import ContextPageCapacityError, StudyMemory
from .study_memory import ContextPageCapacityError, ContextQueryError, StudyMemory
requested_workspace = Path(str(binding.get("memory_workspace", ""))).resolve()
campaign_root = self.campaign.resolve()
try:
Expand All @@ -2708,15 +2728,28 @@ def execute_decision_action(self, episode: Mapping[str, Any], decision: Mapping[
catalog = memory.catalog()
segment_id = catalog.get("current", {}).get("study-public:public-context")
available = set(map(str, binding.get("available_segment_ids", [])))
if not isinstance(segment_id, str) or segment_id not in available:
raise ValueError("Context read must name an available arm-public segment ID")
page_limit = 6000 if decision.get("context_limit") is None else decision.get("context_limit")
measure = context_read_measure
pointer = decision.get("context_pointer") or ""
offset = 0 if decision.get("context_offset") is None else decision.get("context_offset")
try:
if not isinstance(segment_id, str) or segment_id not in available:
raise ContextQueryError("Context read must name an available arm-public segment ID")
page = memory.read_json(segment_id, pointer=pointer, offset=offset,
limit=page_limit, response_capacity=context_response_capacity, measure_response=measure)
except ContextQueryError as exc:
feedback = {"reason": "invalid_context_query", "error": str(exc),
"segment_id": segment_id, "pointer": pointer, "offset": offset, "limit": page_limit,
"available_segment_ids": sorted(available), "root_pointer": "",
"instruction": "Choose a listed public segment and an existing JSON Pointer in it, "
"or use the empty pointer to read its root. '/' addresses an empty-string key, not the root. "
"Use a nonnegative offset and positive page limit."}
return {"action_work_id": work_id, "kind": "analysis", "state": "retryable",
"reason": "invalid_context_query", "request_new_decision": True,
"response_validation_feedback": feedback,
"result": {"operation": "invalid_context_query", "segment_id": segment_id,
"pointer": pointer, "content": None, "delivery_status": "not_delivered",
"observational": True, "scientific_progress_eligible": False}}
except ContextPageCapacityError as exc:
metadata = memory.read_json(segment_id, pointer=pointer, offset=offset, limit=1)
page = {key: metadata[key] for key in ("segment_id", "pointer", "total_characters", "complete")
Expand Down Expand Up @@ -2783,8 +2816,21 @@ def supervisor_work(source_id):
prerequisite = inspect_analysis_prerequisites(binding, registered_actions=bindings,
study_work_lookup=source_work, supervisor_work_lookup=supervisor_work)
if prerequisite is not None:
# Keep the classifier's full, read-only diagnosis in the action
# result and deliver a compact actionable projection as the
# next-turn feedback. Without this bridge request_new_decision
# is true, but the Agent only sees its original proposal and
# cannot tell which registered inference/config to schedule.
feedback = {key: prerequisite[key] for key in (
"status", "classification", "next_executable_operations") if key in prerequisite}
feedback["instruction"] = (
"Use these registered source-work IDs and configuration references to choose "
"the next eligible action. Schedule only work that is not already running; "
"wait for active work or reconcile uncertain receipts. Then retry the same "
"analysis action after its source receipts are valid.")
return {"action_work_id": work_id, "kind": "analysis", "state": "retryable",
"reason": "MISSING_PREREQUISITES", "request_new_decision": True, "result": prerequisite}
"reason": "MISSING_PREREQUISITES", "request_new_decision": True,
"response_validation_feedback": feedback, "result": prerequisite}
if prior and prior["state"] in {"failed", "cancelled", "budget_exhausted"}:
return {"action_work_id": work_id, "kind": binding["kind"], "state": prior["state"],
"result": json.loads(prior["result_json"] or "{}")}
Expand All @@ -2794,16 +2840,43 @@ def supervisor_work(source_id):
(self.campaign_id, str(binding["selection_work_id"]))).fetchone()
target = (binding.get("compared_configuration_ids") or [None, None])[-1]
selected = json.loads(selection["result_json"] or "{}") if selection and selection["state"] == "completed" else {}
def action_selectors(entry):
return {"action_work_id": entry["work_id"],
"kind": entry.get("action_name", entry.get("analysis_operation", action)),
"configuration_id": entry.get("configuration_id"), "seed": entry.get("seed")}
selection_state = selection["state"] if selection else "not_registered"
selection_observation = {"work_id": binding["selection_work_id"], "state": selection_state,
"result": selected}
if not selected:
reason = "source_selection_must_complete_before_comparison"
next_actions = [action_selectors(entry) for entry in bindings if isinstance(entry, Mapping)
and entry.get("work_id") == binding["selection_work_id"]
and entry.get("kind") == "analysis" and entry.get("analysis_operation") == "select"
and entry.get("action_name", "select") == "select"]
return {"action_work_id": work_id, "kind": binding["kind"], "state": "retryable",
"request_new_decision": True, "reason": "source_selection_must_complete_before_comparison",
"selection_work_id": binding["selection_work_id"]}
"request_new_decision": True, "reason": reason,
"selection_work_id": binding["selection_work_id"],
"response_validation_feedback": {"reason": reason,
"attempted_action": action_selectors(binding), "source_selection": selection_observation,
"registered_next_actions": next_actions,
"instruction": "The referenced source selection has no completed result. Inspect its current state and prerequisites, or choose its registered selection action when eligible. A pending or failed selection is not completed evidence. The blocked comparison and later actions in this batch were not executed."}}
if str(selected.get("selected_configuration_id")) != str(target):
reason = "comparison_must_use_source_selected_configuration"
next_actions = [action_selectors(entry) for entry in bindings if isinstance(entry, Mapping)
and entry.get("work_id") and entry.get("kind") == "analysis"
and entry.get("analysis_operation") == "compare" and entry.get("action_name", "compare") == "compare"
and entry.get("selection_work_id") == binding["selection_work_id"]
and str((entry.get("compared_configuration_ids") or [None, None])[-1])
== str(selected.get("selected_configuration_id"))]
return {"action_work_id": work_id, "kind": binding["kind"], "state": "retryable",
"request_new_decision": True, "reason": "comparison_must_use_source_selected_configuration",
"request_new_decision": True, "reason": reason,
"selection_work_id": binding["selection_work_id"],
"source_selected_configuration_id": selected.get("selected_configuration_id"),
"proposed_comparison_configuration_id": target}
"proposed_comparison_configuration_id": target,
"response_validation_feedback": {"reason": reason,
"attempted_action": action_selectors(binding), "source_selection": selection_observation,
"proposed_comparison_configuration_id": target, "registered_next_actions": next_actions,
"instruction": "The referenced source selection is already completed. This comparison targets a different configuration from that saved result. Repeating the same completed selection returns its cached original result, not new evidence. Choose a matching registered comparison below, or inspect the source evidence if none is available. The blocked comparison and later actions in this batch were not executed."}}
self.ledger.register_action(self.campaign_id, work_id, binding, parent_id=str(episode["work_id"]))
if defer_gpu and binding["kind"] in {"fit", "infer"}:
return {"action_work_id": work_id, "kind": binding["kind"], "state": "retryable",
Expand Down
Loading
Loading