1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
//! Server-side execution of a DECLARED command body (`runs command <name>`).
//!
//! The sibling of the string-body path in [`super::declared_body`], and it
//! shares everything that is about the SERVER rather than about the command:
//! the same cancel-path registration, the same live transcript sequencer, the
//! same per-attempt bound enforcement, and the same FFI result contract.
//!
//! What differs is only what is executed. A string body is a command LINE the
//! executor parses here; a declared body arrives already emitted — body
//! lines of templated argv slots, declared environment and working
//! directory — so the server never holds a string that could be re-split,
//! and never has to decide how to split one.
//!
//! Its own file because [`super::declared_body`] is at the workspace's
//! file-size law, and because the two paths' subjects genuinely differ:
//! nothing here parses.
use aion::ActivityDispatch;
use aion_package::DeclaredCommandContract;
use aion_package::contract::CommandBodyCapture;
use aion_worker::shell::{
DeclaredCommandAction, needs_workspace_root, resolve_working_directory, shape_command_result,
};
use super::declared_body::{DeclaredCommandDispatcher, decode_arguments, run_bounded};
use super::workspace_root::WORKSPACE_ROOT_PLACEHOLDER;
impl DeclaredCommandDispatcher {
/// Execute one declared-command attempt and encode the outcome onto the
/// FFI string contract (`retryable:`/`terminal:` on the error side).
pub(super) fn run_declared_command_body(
&self,
request: &ActivityDispatch,
capture: CommandBodyCapture,
command: DeclaredCommandContract,
) -> Result<String, String> {
let arguments = decode_arguments(&request.input)?;
let action = self.declared_command_action(request, command)?;
// The live transcript seam for this attempt. The context owns the
// sending end, so dropping it after the run closes the stream and ends
// the pump — which is then awaited, so no observed line is abandoned
// unpublished when the command finishes.
let (events, drain) = tokio::sync::mpsc::unbounded_channel();
let (context, cancellation) = aion_worker::ActivityContext::with_transcript(
request.workflow_id.clone(),
request.run_id.clone(),
request.activity_id.clone(),
request.attempt,
events,
);
let registration = self.join_cancel_path(request, &cancellation)?;
tracing::info!(
operation = "declared_command_dispatch",
workflow_id = %request.workflow_id,
activity_id = %request.activity_id,
activity_name = %request.name,
task_queue = %request.task_queue,
attempt = request.attempt,
command = action.name(),
"executing a declared command action body at the server"
);
crate::death_note::breadcrumb(&format!(
"declared-command start action={} command={} workflow_id={} run_id={} \
activity_id={} attempt={}",
request.name,
action.name(),
request.workflow_id,
request.run_id,
request.activity_id,
request.attempt,
));
// The bound the DISPATCH authored, or `None` when it authored none —
// and `None` means the command runs to completion however long it
// takes. It is the ONLY bound over a declared command: the command
// declaration itself states no timeout, so what the calling document
// authorised for this attempt is the whole of what limits it.
let bound = aion::activity_timeout_from_config(&request.config);
let transcript = self.transcript();
let ended = self.tokio().block_on(async move {
let pump = tokio::spawn(
super::declared_body_transcript::publish_declared_transcript(transcript, drain),
);
let ended = run_bounded(action.run(&arguments, &context), &cancellation, bound).await;
// Closing the seam is what ends the pump; the context holds it.
drop(context);
if let Err(error) = pump.await {
tracing::warn!(
%error,
operation = "declared_command_dispatch",
"declared command transcript: the publishing task ended abnormally; some \
output lines may not have been retained"
);
}
ended
});
// The command is over and its group is gone, so the attempt leaves the
// cancel path. Dropped explicitly, here and not earlier: while this
// lives, a cancel arriving mid-run still reaches the process.
drop(registration);
let name = request.name.clone();
super::declared_body::encode_end(request, ended, move |outcome| {
let result = shape_command_result(&name, capture, outcome)?;
serde_json::to_string(&result).map_err(|error| {
aion_worker::ActivityFailure::terminal(format!(
"declared command result failed to encode: {error}"
))
})
})
}
/// Build the executor's action, with the server-resolved working directory
/// already in place.
///
/// Ratification condition (#139), unchanged in substance from the string
/// path: a body that USES the placeholder is refused terminally, by name,
/// when the root cannot resolve to an absolute directory that exists — no
/// fallback to cwd, temp, or anything else.
///
/// The splice itself is [`aion_worker::shell::resolve_working_directory`],
/// which is also what `aion worker awl` calls. The server contributes only
/// the answer to "where is the root" — its aion home — and the rules over
/// that answer are shared, so the two executors cannot disagree about the
/// world a process runs in.
///
/// The argv-word rules the STRING path imposes (`WorkspaceRoot::expand`,
/// which requires each occurrence to stand as a whole unquoted word) do
/// not apply: a declared `cwd` is one whole path that nothing splits, so
/// imposing them would refuse paths that are perfectly well formed here.
fn declared_command_action(
&self,
request: &ActivityDispatch,
command: DeclaredCommandContract,
) -> Result<DeclaredCommandAction, String> {
let action = DeclaredCommandAction::new(command);
let Some(declared) = action.declared_working_directory().map(str::to_owned) else {
return Ok(action);
};
// A cwd that never names the placeholder never reaches the root, so a
// server whose home could not resolve still serves every body that did
// not ask where the workspace is.
if !needs_workspace_root(&declared) {
return Ok(action.with_working_directory(declared));
}
let refuse = |error: &dyn std::fmt::Display| {
format!(
"terminal:declared command for action `{name}` states a working directory \
using the {placeholder} placeholder and cannot dispatch: {error}",
name = request.name,
placeholder = WORKSPACE_ROOT_PLACEHOLDER,
)
};
let root = self.workspace_root().resolved().map_err(|error| {
// A root that never resolved is carried rather than raised at
// boot, so this is where it surfaces: at the first dispatch that
// actually needed it, naming the action and the reason.
refuse(error)
})?;
let directory =
resolve_working_directory(&declared, root).map_err(|error| refuse(&error))?;
tracing::info!(
operation = "declared_command_dispatch",
workflow_id = %request.workflow_id,
activity_id = %request.activity_id,
activity_name = %request.name,
task_queue = %request.task_queue,
attempt = request.attempt,
working_directory = %directory.display(),
"resolved the declared command's working directory against this server's root"
);
Ok(action.with_working_directory(directory))
}
}
#[cfg(test)]
#[path = "declared_command_body_tests.rs"]
mod tests;