{body}\n'
+
+
+app.format("html", render=_html_page, media_type="text/html")
+
+
class Payload:
"""A command's flat payload, written as ``data`` under its declared keys.
``command`` names the command in the operating contract; its output
shape is this payload's schema. A key the shape marks optional is
- written as an empty object when the command has nothing for it, since
+ written as ``null`` when the command has nothing for it, since
treaty writes every key of a schema on every answer.
"""
command: ClassVar[str]
+ writes: ClassVar[bool] = False
+ """The command changes state, so the payload carries ``effect`` (REQ-C-003)."""
def __init__(self, body: Mapping[str, object]) -> None:
"""Keep the payload as the domain serialized it."""
@@ -247,12 +303,32 @@ def _shape_keys(command: str) -> tuple[tuple[str, bool], ...]:
for entry in CLI_COMMANDS
if entry.program == "cloudfall" and entry.name == command
)
- (shape,) = contract.output
- return tuple((key.name, key.optional) for key in shape.keys if key.name != "error")
+ # A command with two shapes (a plan without --yes, a result with it)
+ # writes one object: a key not in every shape is optional, so null.
+ names: list[str] = []
+ for shape in contract.output:
+ names.extend(
+ key.name
+ for key in shape.keys
+ if key.name != "error" and key.name not in names
+ )
+ return tuple(
+ (
+ name,
+ any(
+ all(key.name != name for key in shape.keys)
+ or any(key.name == name and key.optional for key in shape.keys)
+ for shape in contract.output
+ ),
+ )
+ for name in names
+ )
def _payload_schema(cls: type[Payload]) -> dict[str, object]:
names = [name for name, _ in _shape_keys(cls.command)]
+ if cls.writes:
+ names.append("effect")
return {
"type": "object",
# The domain writes each array in the order it means: a ranking, a
@@ -267,7 +343,7 @@ def _payload_document(payload: Payload) -> dict[str, object]:
document = dict(payload.body)
for name, optional in _shape_keys(payload.command):
if optional and name not in document:
- document[name] = {}
+ document[name] = None
return document
@@ -334,6 +410,8 @@ class Fleet:
directory: Path
state: ValidatedConfig
read: FleetRead | None = None
+ source: InventorySource | None = None
+ """The team's Ansible inventory the fleet was read from, if any."""
@classmethod
def acquire(cls, args: ProjectArgs, _ctx: Ctx) -> Self:
@@ -342,7 +420,7 @@ def acquire(cls, args: ProjectArgs, _ctx: Ctx) -> Self:
source = _inventory_source(args)
if source is not None:
read = read_fleet(read_inventory(source), args.schemas)
- return cls(Path.cwd(), read.config, read)
+ return cls(Path.cwd(), read.config, read, source)
directory = resolve_project_directory(
args.project, os.environ, Path.cwd()
)
@@ -432,7 +510,7 @@ class InventoryPayload(Payload):
"show",
description=(
"Show the typed, secret-free platform inventory; data.ansible says "
- "how it was read from an Ansible inventory, empty for a project"
+ "how it was read from an Ansible inventory, null for a project"
),
danger_level="safe",
exit_codes=[
@@ -656,7 +734,7 @@ class ProposalPayload(Payload):
"list",
description="List proposal receipts, oldest first",
danger_level="safe",
- exit_codes=["PROJECT_INVALID", "CONFIG_INVALID", "RECORD_INVALID"],
+ exit_codes=["PROJECT_INVALID", "CONFIG_INVALID", "RECORD_INVALID", "NOT_FOUND"],
examples=[("List the operator's proposals", "cloudfall operator list")],
)
def operator_list(args: ProposalArgs, _ctx: Ctx, fleet: Fleet) -> ProposalsPayload:
@@ -1092,6 +1170,7 @@ def _authoring_failed(error: AuthoringError) -> Exception:
"CONFLICT",
"PERMISSION_DENIED",
"PRECONDITION",
+ "NOT_FOUND",
]
@@ -1101,7 +1180,7 @@ def _authoring_failed(error: AuthoringError) -> Exception:
danger_level="mutating",
timeout=30,
supports_raw_payload=True,
- exit_codes=[*_ADD_EXIT_CODES, "NOT_FOUND"],
+ exit_codes=_ADD_EXIT_CODES,
examples=[
(
"Declare your own key",
@@ -1440,6 +1519,7 @@ class HealthPayload(Payload):
"PROJECT_INVALID",
"CONFIG_INVALID",
"PRECONDITION",
+ "ENGINE_STEP_FAILED",
"UNHEALTHY",
],
timeout=None,
@@ -2090,3 +2170,758 @@ def migrate(args: MigrateArgs, ctx: Ctx, fleet: Fleet) -> Migration:
message = f"the migration failed at {migration.step}"
raise Exit.ENGINE_STEP_FAILED(message, context=cause, data=migration)
return migration
+
+
+# why
+
+
+@dataclass(frozen=True, slots=True, kw_only=True)
+class WhyArgs(DecisionsArgs):
+ """Arguments of ``why``."""
+
+ host: str | None = Flag(
+ default=None, description="Only decisions whose record names this host"
+ )
+ operation: ResourceId | None = Flag(
+ default=None, description="Only decisions of this declared operation"
+ )
+ since: str | None = Flag(
+ default=None,
+ description=(
+ "Only decisions with a moment at or after this ISO 8601 time or date"
+ ),
+ )
+ until: str | None = Flag(
+ default=None,
+ description=(
+ "Only decisions with a moment at or before this ISO 8601 time or date"
+ ),
+ )
+
+ def __post_init__(self) -> None:
+ """Refuse a time the question cannot be asked with."""
+ DecisionsArgs.__post_init__(self)
+ try:
+ self.query()
+ except WhyError as error:
+ raise ParseError(error.detail, context={"code": error.code}) from error
+
+ def query(self) -> WhyQuery:
+ """Return the question, as the record is filtered by it."""
+ return WhyQuery.from_boundary(
+ host=self.host,
+ operation=self.operation.value if self.operation is not None else None,
+ since=self.since,
+ until=self.until,
+ )
+
+
+class WhyPayload(Payload):
+ """``why``: each decision the question is about, told from its record."""
+
+ command = "why"
+
+
+@app.command(
+ "why",
+ description=(
+ "Answer why the agent did that, from the record, for a host, an operation "
+ "or a time window; --format html renders one page"
+ ),
+ danger_level="safe",
+ exit_codes=["RECORD_INVALID"],
+ renderers={"html": render_why_document},
+ examples=[
+ ("Ask about one host", "cloudfall why --host h1"),
+ ("One page for a person", "cloudfall why --since 2026-10-01 --format html"),
+ ],
+)
+def why(args: WhyArgs, _ctx: Ctx) -> WhyPayload:
+ """Answer from the record alone: no catalog, no fleet."""
+ store = DecisionStore(
+ directory=args.root / args.decisions, catalog=SchemaCatalog(args.schemas)
+ )
+ try:
+ result = answer(store, args.query())
+ except (DecisionError, WhyError) as error:
+ raise Exit.RECORD_INVALID(error.detail, context={"code": error.code}) from error
+ return WhyPayload(result.as_dict())
+
+
+# observe, secrets render, import render, import render-api: payloads with
+# camelCase keys, written through the adapter now that it may carry effect.
+
+
+app.exit_code(
+ "INCOMPLETE",
+ 91,
+ description=(
+ "Some servers produced no snapshot or Ansible failed; data says which"
+ ),
+ retryable=True,
+ # The hosts are only read; a retry rewrites the snapshots it wrote.
+ side_effects="none",
+ suggestion="read data.missing and data.detail, fix the hosts, then observe again",
+)
+
+
+@dataclass(frozen=True, slots=True, kw_only=True)
+class ObserveArgs(FleetArgs):
+ """Arguments of ``observe``."""
+
+ output_dir: Path = Flag(
+ default=Path("tmp/observed"),
+ description="Snapshot directory to write (default: tmp/observed)",
+ )
+ limit: str | None = Flag(
+ default=None, description="Ansible host pattern to inspect a subset"
+ )
+ engine: Path = Flag(
+ default=default_engine_directory(),
+ description="Engine directory holding the playbooks (default: bundled)",
+ )
+
+ def __post_init__(self) -> None:
+ """Keep a relative output path inside the project."""
+ _inside_project(self.output_dir, "output-dir")
+
+
+class ObservedPayload(Payload):
+ """``observe``: which servers produced a snapshot."""
+
+ command = "observe"
+ writes = True
+
+
+def _playbook_runner(
+ ctx: Ctx, directory: Path
+) -> Callable[[Sequence[str], Mapping[str, str]], PlaybookRun]:
+ """Run the inspection through ``ctx.run``, keeping its output for a failure."""
+
+ def run(argv: Sequence[str], environment: Mapping[str, str]) -> PlaybookRun:
+ done = ctx.run(list(argv), env=environment, cwd=directory, check=False)
+ return PlaybookRun(exit_code=done.returncode, output=done.stdout + done.stderr)
+
+ return run
+
+
+@app.command(
+ "observe",
+ description=(
+ "Collect read-only server snapshots from the fleet; servers without one "
+ "exit non-zero with the run in data"
+ ),
+ danger_level="mutating",
+ exit_codes=[
+ "PROJECT_INVALID",
+ "CONFIG_INVALID",
+ "INVENTORY_UNREADABLE",
+ "PRECONDITION",
+ "INCOMPLETE",
+ ],
+ requires=[ONE_FLEET],
+ timeout=None,
+ subprocess=Subprocess("ansible-playbook"),
+ required_tools={"ansible-playbook": "2.21.0"},
+ supports_raw_payload=True,
+ examples=[("Snapshot every server", "cloudfall observe")],
+)
+def observe(args: ObserveArgs, ctx: Ctx, fleet: Fleet) -> ObservedPayload:
+ """Inspect every declared server and write one snapshot each."""
+ sources = (
+ (fleet.source.value,)
+ if fleet.source is not None
+ else (fleet.directory / "tmp/ansible-inventory.json",)
+ )
+ output = fleet.path(args.output_dir)
+ effect = "updated" if output.exists() else "created"
+ try:
+ request = ObservationRequest(
+ inventory_sources=sources,
+ output_directory=output.resolve(),
+ engine_directory=args.engine,
+ limit=args.limit,
+ configuration=(
+ team_configuration(Path.cwd()) if fleet.source is not None else None
+ ),
+ )
+ result = collect_observations(
+ fleet.inventory,
+ request,
+ fleet.path(Path("tmp/cloudfall")),
+ _playbook_runner(ctx, fleet.directory),
+ )
+ except ObserveError as error:
+ raise Exit.PRECONDITION(error.detail, context={"code": error.code}) from error
+ payload = ObservedPayload({**result.as_dict(), "effect": effect})
+ if not result.complete:
+ message = f"{len(result.missing)} server(s) produced no snapshot"
+ raise Exit.INCOMPLETE(message, data=payload)
+ return payload
+
+
+# secrets render
+
+
+secrets = app.group("secrets", description="Resolve declared secret references")
+
+
+@dataclass(frozen=True, slots=True, kw_only=True)
+class SecretsRenderArgs(EngineArgs):
+ """Arguments of ``secrets render``."""
+
+ component: ResourceId = Arg(description="Component whose references to render")
+ secrets_dir: Path = Flag(
+ default=Path("secrets"),
+ description="sops-encrypted secrets directory (default: secrets)",
+ secret=False,
+ )
+ output_file: Path | None = Flag(
+ default=None,
+ description="Environment file to write (default: tmp/env/The record holds no decision this question is about.
' ) filters = ", ".join( f"{escape(key)} = {escape(str(value))}" - for key, value in result.query.as_dict().items() + for key, value in query.items() ) asked = escape(filters) if filters else "Every decision the record holds" - count = len(result.explanations) + count = len(answers) counted = f"{count} decision{'' if count == 1 else 's'}" return f""" @@ -285,7 +296,7 @@ def render_why_html(result: WhyAnswer) -> str:{asked} · {counted}
- · from {escape(result.directory)}
{escape(str(document["directory"]))}