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
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
use super::chain::log_tx;
use alloy_primitives::Bytes;
use blueprint_client_tangle::TangleClient;
use cargo_tangle::command::jobs::check::wait_for_job_result;
use cargo_tangle::command::jobs::helpers::{
JobSchema, list_jobs, load_job_call_details, load_job_schema, print_job_call_details,
print_job_summaries,
};
use cargo_tangle::command::jobs::submit::submit_job as submit_job_call;
use cargo_tangle::command::tangle::TangleClientArgs;
use clap::Subcommand;
use color_eyre::eyre::{Context, Result, eyre};
use serde_json::json;
use std::{fs, path::PathBuf, time::Duration};
#[derive(Subcommand, Debug)]
pub(super) enum JobsCommands {
/// List all jobs defined by a blueprint.
///
/// Shows job indices, names, input schemas, and output types.
List {
#[command(flatten)]
network: TangleClientArgs,
/// Blueprint ID to list jobs for.
#[arg(long)]
blueprint_id: u64,
/// Output as JSON instead of formatted table.
#[arg(long)]
json: bool,
},
/// Show details for a submitted job call.
///
/// Displays inputs, outputs, status, and result data.
Show {
#[command(flatten)]
network: TangleClientArgs,
/// Blueprint ID the service belongs to.
#[arg(long)]
blueprint_id: u64,
/// Service ID the job was submitted to.
#[arg(long)]
service_id: u64,
/// Call ID returned when the job was submitted.
#[arg(long)]
call_id: u64,
/// Output as JSON instead of formatted display.
#[arg(long)]
json: bool,
},
/// Submit a job to a running service.
///
/// Invokes a job on the service operators. Inputs can be provided as:
/// - Raw hex bytes (--payload-hex)
/// - Binary file (--payload-file)
/// - Structured JSON matching the job schema (--params-file)
/// - Interactive prompts (--prompt)
Submit {
#[command(flatten)]
network: TangleClientArgs,
/// Blueprint ID the service belongs to.
#[arg(long)]
blueprint_id: u64,
/// Service ID to submit the job to.
#[arg(long)]
service_id: u64,
/// Job index (0-based) as defined in the blueprint.
#[arg(long)]
job: u8,
/// Job inputs as hex-encoded bytes (without 0x prefix).
#[arg(long = "payload-hex", value_name = "HEX")]
payload_hex: Option<String>,
/// File containing raw job input bytes.
#[arg(long = "payload-file", value_name = "FILE")]
payload_file: Option<PathBuf>,
/// JSON file with structured inputs matching the job schema.
#[arg(
long = "params-file",
value_name = "FILE",
conflicts_with_all = ["payload_hex", "payload_file"]
)]
params_file: Option<PathBuf>,
/// Interactively prompt for each job input.
#[arg(
long,
conflicts_with_all = ["payload_hex", "payload_file", "params_file"],
action = clap::ArgAction::SetTrue
)]
prompt: bool,
/// Wait for job result after submission.
#[arg(long)]
watch: bool,
/// Timeout in seconds when watching for result.
#[arg(long, default_value_t = 60)]
timeout_secs: u64,
/// Output transaction details as JSON.
#[arg(long)]
json: bool,
},
/// Wait for a job result by call ID.
///
/// Polls for job completion and displays the result when available.
Watch {
#[command(flatten)]
network: TangleClientArgs,
/// Blueprint ID the service belongs to.
#[arg(long)]
blueprint_id: u64,
/// Service ID the job was submitted to.
#[arg(long)]
service_id: u64,
/// Call ID from the job submission.
#[arg(long)]
call_id: u64,
/// Timeout in seconds before giving up.
#[arg(long, default_value_t = 60)]
timeout_secs: u64,
},
}
pub(super) async fn run(command: JobsCommands) -> Result<()> {
match command {
JobsCommands::List {
network,
blueprint_id,
json,
} => {
let client = network.connect(blueprint_id, None).await?;
let jobs = list_jobs(&client, blueprint_id).await?;
print_job_summaries(&jobs, json);
}
JobsCommands::Show {
network,
blueprint_id,
service_id,
call_id,
json,
} => {
let client = network.connect(blueprint_id, Some(service_id)).await?;
let details = load_job_call_details(&client, blueprint_id, service_id, call_id).await?;
print_job_call_details(&details, json);
}
JobsCommands::Submit {
network,
blueprint_id,
service_id,
job,
payload_hex,
payload_file,
params_file,
prompt,
watch,
timeout_secs,
json,
} => {
let client = network.connect(blueprint_id, Some(service_id)).await?;
let mut schema_cache: Option<JobSchema> = None;
let payload = match (payload_hex, payload_file, params_file, prompt) {
(Some(hex_value), None, None, false) => decode_payload_hex(&hex_value)?,
(None, Some(path), None, false) => read_payload_file(&path)?,
(None, None, Some(path), false) => {
ensure_schema_loaded(&mut schema_cache, &client, blueprint_id, job).await?;
schema_cache
.as_ref()
.expect("schema should be loaded")
.encode_params_from_file(&path)?
}
(None, None, None, true) => {
ensure_schema_loaded(&mut schema_cache, &client, blueprint_id, job).await?;
schema_cache
.as_ref()
.expect("schema should be loaded")
.prompt_for_params()?
}
(Some(_), Some(_), _, _) => {
return Err(eyre!(
"Specify only one of --payload-hex, --payload-file, --params-file, or --prompt"
));
}
_ => {
return Err(eyre!(
"Provide job inputs via --payload-hex, --payload-file, --params-file, or --prompt"
));
}
};
let submission = submit_job_call(&client, service_id, job, payload.clone()).await?;
log_tx("Job submission", &submission.tx, json);
if json {
println!(
"{}",
json!({
"event": "job_submitted",
"service_id": service_id,
"blueprint_id": blueprint_id,
"job": job,
"call_id": submission.call_id,
"tx_hash": format!("{:#x}", submission.tx.tx_hash),
})
);
} else {
println!(
"Submitted job {job} to service {service_id}. Call ID: {} (tx: {:#x})",
submission.call_id, submission.tx.tx_hash
);
}
if watch {
let bytes = wait_for_job_result(
&client,
service_id,
submission.call_id,
Duration::from_secs(timeout_secs),
)
.await?;
match ensure_schema_loaded(&mut schema_cache, &client, blueprint_id, job).await {
Ok(()) => {
let schema = schema_cache.as_ref().expect("schema present");
match schema.decode_and_format_results(&bytes) {
Ok(Some(lines)) => {
if json {
println!(
"{}",
json!({
"event": "job_result",
"service_id": service_id,
"call_id": submission.call_id,
"decoded": lines,
"length": bytes.len(),
})
);
} else {
println!(
"Job result ready ({} bytes). Decoded output:",
bytes.len()
);
for line in lines {
println!(" {line}");
}
}
}
Ok(None) => print_raw_job_result(&bytes),
Err(err) => {
eprintln!("Failed to decode job result: {err}");
print_raw_job_result(&bytes);
}
}
}
Err(err) => {
eprintln!("Unable to load job schema: {err}");
print_raw_job_result(&bytes);
}
}
}
}
JobsCommands::Watch {
network,
blueprint_id,
service_id,
call_id,
timeout_secs,
} => {
let client = network.connect(blueprint_id, Some(service_id)).await?;
let bytes = wait_for_job_result(
&client,
service_id,
call_id,
Duration::from_secs(timeout_secs),
)
.await?;
print_raw_job_result(&bytes);
}
}
Ok(())
}
async fn ensure_schema_loaded(
cache: &mut Option<JobSchema>,
client: &TangleClient,
blueprint_id: u64,
job_index: u8,
) -> Result<()> {
if cache.is_none() {
let schema = load_job_schema(client, blueprint_id, job_index).await?;
*cache = Some(schema);
}
Ok(())
}
fn decode_payload_hex(value: &str) -> Result<Bytes> {
let trimmed = value.trim();
let raw = trimmed.trim_start_matches("0x");
let bytes = hex::decode(raw).context("invalid payload hex")?;
Ok(Bytes::from(bytes))
}
fn read_payload_file(path: &PathBuf) -> Result<Bytes> {
let data = fs::read(path)
.with_context(|| format!("Failed to read payload file {}", path.display()))?;
Ok(Bytes::from(data))
}
fn print_raw_job_result(bytes: &[u8]) {
println!(
"Job result ready ({} bytes): 0x{}",
bytes.len(),
hex::encode(bytes)
);
}