owenisas commited on
Commit
ca0e2ae
·
verified ·
1 Parent(s): 22f9ecd

Bound automation preflight waits by the run deadline

Browse files
Files changed (2) hide show
  1. README.md +2 -0
  2. client.py +44 -6
README.md CHANGED
@@ -80,6 +80,8 @@ Input JSON:
80
 
81
  The client requires an explicitly supplied inference-token environment variable and the existing prepaid-overage-disabled operator attestation. It does not use the saved deployment owner's token for generation. See `python client.py --help` for the current CLI contract. No paid fallback is configured.
82
 
 
 
83
  ## Runtime configuration
84
 
85
  Default startup serves the UI and non-inference APIs without downloading weights.
 
80
 
81
  The client requires an explicitly supplied inference-token environment variable and the existing prepaid-overage-disabled operator attestation. It does not use the saved deployment owner's token for generation. See `python client.py --help` for the current CLI contract. No paid fallback is configured.
82
 
83
+ `run --max-wall-seconds` applies one deadline across caller/revision checks, client initialization, health/planning waits and inference waits. Timed-out preflight reads cannot advance to GPU submission; late-created clients close. Local staging, journal writes and bounded cleanup add overhead. A post-submission timeout remains an unknown outcome, not permission to retry.
84
+
85
  ## Runtime configuration
86
 
87
  Default startup serves the UI and non-inference APIs without downloading weights.
client.py CHANGED
@@ -18,6 +18,7 @@ there is deliberately no resume, retry, cancellation, or token rotation.
18
  from __future__ import annotations
19
 
20
  import argparse
 
21
  from contextlib import contextmanager, redirect_stderr, redirect_stdout, suppress
22
  from dataclasses import dataclass
23
  import hashlib
@@ -318,6 +319,39 @@ def _new_client(factory, src, token, downloads=False):
318
  httpx_kwargs={"timeout": 20})
319
 
320
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
321
  def _capture(hop, client, job=None):
322
  session = getattr(client, "session_hash", None)
323
  if isinstance(session, str):
@@ -418,7 +452,7 @@ def _run(args, request, token, factory, metadata, identity, wrap, clock):
418
  journal.data["request"] = {"preset_id": request.preset_id, "prompt": request.prompt,
419
  "canvas": request.canvas, "duration": request.duration, "seed": request.seed}
420
  journal.save()
421
- caller = identity(token).get("name")
422
  if not isinstance(caller, str) or not caller:
423
  raise AuthenticationRequired()
424
  if caller.casefold() == "owenisas":
@@ -426,24 +460,28 @@ def _run(args, request, token, factory, metadata, identity, wrap, clock):
426
  journal.data["caller"] = caller
427
  for repo, revision in ((GENERATOR_REPO, args.expected_revision),
428
  (CONDITIONER_REPO, CONDITIONER_REVISION)):
429
- if metadata(repo, token).get("sha") != revision:
430
  raise RevisionMismatch()
431
  downloads = {}
432
  for name in ("generator", "conditioner"):
433
  downloads[name] = journal.directory / (name + "-downloads")
434
  downloads[name].mkdir(mode=0o700)
435
- generator = _new_client(factory, GENERATOR, token, downloads["generator"])
 
 
436
  clients.append(generator)
437
- health = generator.predict(api_name="/health")
438
  if (not isinstance(health, dict) or health.get("status") != "ready"
439
  or health.get("model_loaded") is not True or health.get("blockers") != []
440
  or health.get("conditioner_revision") != CONDITIONER_REVISION):
441
  raise RuntimeNotReady()
442
  journal.data["health"] = health
443
- plan = _plan(generator, request)
444
  journal.data["plan"] = plan
445
  journal.save()
446
- conditioner = _new_client(factory, CONDITIONER, token, downloads["conditioner"])
 
 
447
  clients.append(conditioner)
448
 
449
  def image(name):
 
18
  from __future__ import annotations
19
 
20
  import argparse
21
+ from concurrent.futures import Future
22
  from contextlib import contextmanager, redirect_stderr, redirect_stdout, suppress
23
  from dataclasses import dataclass
24
  import hashlib
 
319
  httpx_kwargs={"timeout": 20})
320
 
321
 
322
+ def _within_deadline(call, deadline, clock, cleanup=None):
323
+ """Bound a read-only preflight operation, never an inference submission.
324
+
325
+ SDK predict/initialization may wait beyond an HTTP request timeout. The
326
+ daemon performs only this one read; it cannot advance to a GPU submission
327
+ or update the journal after the caller times out. Late-created clients close.
328
+ """
329
+ if clock() >= deadline:
330
+ raise TimeoutError()
331
+ future = Future()
332
+
333
+ def invoke():
334
+ try:
335
+ future.set_result(call())
336
+ except BaseException as error:
337
+ future.set_exception(error)
338
+
339
+ threading.Thread(target=invoke, name="workflow-preflight", daemon=True).start()
340
+ try:
341
+ remaining = deadline - clock()
342
+ if remaining <= 0:
343
+ raise TimeoutError()
344
+ return future.result(timeout=remaining)
345
+ except BaseException:
346
+ if cleanup is not None:
347
+ def close_late(done):
348
+ with suppress(Exception):
349
+ if not done.cancelled() and done.exception() is None:
350
+ cleanup(done.result())
351
+ future.add_done_callback(close_late)
352
+ raise
353
+
354
+
355
  def _capture(hop, client, job=None):
356
  session = getattr(client, "session_hash", None)
357
  if isinstance(session, str):
 
452
  journal.data["request"] = {"preset_id": request.preset_id, "prompt": request.prompt,
453
  "canvas": request.canvas, "duration": request.duration, "seed": request.seed}
454
  journal.save()
455
+ caller = _within_deadline(lambda: identity(token), deadline, clock).get("name")
456
  if not isinstance(caller, str) or not caller:
457
  raise AuthenticationRequired()
458
  if caller.casefold() == "owenisas":
 
460
  journal.data["caller"] = caller
461
  for repo, revision in ((GENERATOR_REPO, args.expected_revision),
462
  (CONDITIONER_REPO, CONDITIONER_REVISION)):
463
+ if _within_deadline(lambda repo=repo: metadata(repo, token), deadline, clock).get("sha") != revision:
464
  raise RevisionMismatch()
465
  downloads = {}
466
  for name in ("generator", "conditioner"):
467
  downloads[name] = journal.directory / (name + "-downloads")
468
  downloads[name].mkdir(mode=0o700)
469
+ generator = _within_deadline(
470
+ lambda: _new_client(factory, GENERATOR, token, downloads["generator"]),
471
+ deadline, clock, cleanup=lambda client: client.close())
472
  clients.append(generator)
473
+ health = _within_deadline(lambda: generator.predict(api_name="/health"), deadline, clock)
474
  if (not isinstance(health, dict) or health.get("status") != "ready"
475
  or health.get("model_loaded") is not True or health.get("blockers") != []
476
  or health.get("conditioner_revision") != CONDITIONER_REVISION):
477
  raise RuntimeNotReady()
478
  journal.data["health"] = health
479
+ plan = _within_deadline(lambda: _plan(generator, request), deadline, clock)
480
  journal.data["plan"] = plan
481
  journal.save()
482
+ conditioner = _within_deadline(
483
+ lambda: _new_client(factory, CONDITIONER, token, downloads["conditioner"]),
484
+ deadline, clock, cleanup=lambda client: client.close())
485
  clients.append(conditioner)
486
 
487
  def image(name):