Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
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
13 changes: 13 additions & 0 deletions examples/autonomous-replan-obligation-smoke.py
Original file line number Diff line number Diff line change
Expand Up @@ -296,6 +296,19 @@ def assert_typed_repeat_requires_two_equivalent_observations() -> None:
assert obligation["replan_context"]["delivery"] == "host_projected", guard
assert guard["replan_action_packet"]["obligation_id"] == obligation["obligation_id"], guard
assert guard["replan_action_packet"]["required_outcome"] == "semantic_delta", guard
cli_actions = guard["interaction_contract"]["cli_channel"]["next_cli_actions"]
replan_action = next(
action for action in cli_actions if "refresh-state" in action
)
# The printed transition stays minimal and always legal: the workspace
# path is rejected for a non-delivery settlement and the replan ACK is
# the agent's own act. Both conditions are carried as compact codes
# beside the obligation instead of growing the agent-facing rows.
assert "--autonomous-replan-recorded" not in replan_action, replan_action
assert "--delivery-workspace-path" not in replan_action, replan_action
assert obligation["settlement_flag_conditions"] == (
"workspace_path:todo_requires,replan_ack:after_replan_recorded"
), obligation


def assert_equivalent_observation_is_rejected_before_write() -> None:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,12 @@
" Use the host-projected coverage ledger and produce a typed semantic delta; "
"repeated observations cannot close replan."
)
# The two flags a printed replan transition cannot carry itself. Kept as compact
# machine codes because the agent-facing quota rows allow only tens of extra
# characters per row, and prose here would exceed that budget.
REPLAN_SETTLEMENT_FLAG_CONDITIONS = (
"workspace_path:todo_requires,replan_ack:after_replan_recorded"
)


def build_replan_novelty_policy() -> dict[str, str]:
Expand Down Expand Up @@ -211,6 +217,7 @@ def ensure_replan_novelty_policy(
str(normalized.get("recommended_action") or "run a bounded autonomous replan")
)
normalized["replan_novelty_policy"] = build_replan_novelty_policy()
normalized["settlement_flag_conditions"] = REPLAN_SETTLEMENT_FLAG_CONDITIONS
rearmed_after_obligation_id = normalize_todo_replan_obligation_id(
normalized.get("rearmed_after_obligation_id")
)
Expand Down