PyHydroGeophysX.agents.runtime package

Submodules

PyHydroGeophysX.agents.runtime.adapters module

A generated reader for a text table nobody anticipated, run only if approved.

The recurring failure this exists for: a coordinate file - electrodes, geophone positions, a topography profile - that holds the right numbers in a layout the readers were not written for. A comma-separated file with two header lines, a station column where none was expected, an easting/northing/elevation triple where an x/z pair was. The data are fine; the parser is not. A person would look at the first few lines and write four lines of Python.

So can a model, and this lets it - under conditions that are the whole point of the module:

The user reads the code and approves it. Every run of generated code is put to RunContext.ask() with the source in the question, and runs only on an explicit yes. Without someone to ask, the answer is no: a headless run declines rather than taking a default, because “nobody was there” must not read as consent to execute code.

It never touches the filesystem, the network or the process. The function is handed the file’s text, not its path, and runs in a namespace with a small builtins whitelist and no imports beyond numpy. It returns numbers.

Its output must look like what it claimed to produce: a non-empty 2-D array of finite numbers with the expected column count, or it is rejected.

It is written down. The source, the approval and the file it produced are recorded as warnings, because a number that reached a report through code a model wrote must be traceable to that fact.

This is a guard rail, not a security boundary. A restricted exec namespace stops the obvious reaches - open, __import__, subprocess - and a determined escape through Python’s object graph is not something it claims to prevent. What actually stands between a user and bad code here is that they read it and pressed the button.

PyHydroGeophysX.agents.runtime.adapters.ADAPTABLE_FILES = {'electrode_file': 3, 'geophone_file': 3, 'topography_file': 2}

plain numeric text tables, where “the layout was unexpected” is the whole failure and the repaired form is unambiguous. Survey data files are absent on purpose - an instrument format is not a table, and a generated parser for one would be guessing at physics rather than at columns.

Type:

Configuration keys an adapter may rewrite

PyHydroGeophysX.agents.runtime.adapters.ADAPTER_RETRY = '\n\nA reader you wrote for this file was approved and run, and it failed with:\n{failure}\nThat reader was:\n{previous}\nFix the cause the error shows.'

Added to ADAPTER_PROMPT when a reader the user approved failed.

PyHydroGeophysX.agents.runtime.adapters.FORBIDDEN = ('import', 'open(', 'exec', 'eval', 'compile', '__', 'globals', 'locals', 'getattr', 'setattr', 'delattr', 'vars', 'input', 'breakpoint', 'exit', 'quit', 'help', 'memoryview')

Names that must not appear in generated source. Checked before it is compiled, so a proposal that reaches for them is never shown to the user as runnable.

PyHydroGeophysX.agents.runtime.adapters.SAFE_BUILTINS = {'IndexError': <class 'IndexError'>, 'StopIteration': <class 'StopIteration'>, 'TypeError': <class 'TypeError'>, 'ValueError': <class 'ValueError'>, 'abs': <built-in function abs>, 'all': <built-in function all>, 'any': <built-in function any>, 'bool': <class 'bool'>, 'dict': <class 'dict'>, 'divmod': <built-in function divmod>, 'enumerate': <class 'enumerate'>, 'filter': <class 'filter'>, 'float': <class 'float'>, 'int': <class 'int'>, 'isinstance': <built-in function isinstance>, 'len': <built-in function len>, 'list': <class 'list'>, 'map': <class 'map'>, 'max': <built-in function max>, 'min': <built-in function min>, 'range': <class 'range'>, 'reversed': <class 'reversed'>, 'round': <built-in function round>, 'set': <class 'set'>, 'sorted': <built-in function sorted>, 'str': <class 'str'>, 'sum': <built-in function sum>, 'tuple': <class 'tuple'>, 'zip': <class 'zip'>}

Everything the generated function is allowed to call.

PyHydroGeophysX.agents.runtime.adapters.TIMEOUT_SECONDS = 5.0

How long a generated function may run. It parses a few hundred lines of text; anything longer is a loop that will not end.

PyHydroGeophysX.agents.runtime.adapters.adapt_file(ctx: RunContext, key: str, error: str, ask: Callable[[str], str] | None) → str

Offer a generated reader for ctx.config[key], and use it if approved.

Returns:

The path of the rewritten file, now a plain numeric table the ordinary readers accept, or “” when nothing was produced. The rewritten file is inside the run’s output directory; the user’s own file is never touched.

Return type:

str

PyHydroGeophysX.agents.runtime.adapters.check_source(code: str) → str

Why code may not be run, or “” when nothing objects.

Examples

>>> check_source('def adapt(text):\n    return [[1.0, 2.0]]')
''
>>> check_source('import os\ndef adapt(text): return []')
"generated code may not use 'import'"
>>> check_source('x = 1')
'generated code must define a function called adapt'

PyHydroGeophysX.agents.runtime.adapters.propose(ctx: RunContext, key: str, path: Any, error: str, ask: Callable[[str], str] | None, previous: str = '', failure: str = '') → Tuple[str, str]

Ask the model for a reader for this file. Returns (code, description).

With previous and failure, the reader it wrote before and how that failed, for it to fix. Returns ("", "") when there is no model, the reply is unusable, or the code it proposed is not allowed to run.

PyHydroGeophysX.agents.runtime.adapters.review(ctx: RunContext, key: str, path: Any, code: str, description: str) → bool

Show the user the code and ask whether to run it.

The answer is no unless somebody says yes. A run with nobody to ask declines - RunContext.ask() returns its default, and the default here is “do not run it”, because silence is not consent to execute code.

PyHydroGeophysX.agents.runtime.adapters.run_adapter(code: str, text: str, columns: int, timeout: float = 5.0) → ndarray

Compile and run code’s adapt on text, and check what came back.

Parameters:

• code (str) – The generated source. Must already have passed check_source().

• text (str) – The file’s contents, passed as a value - the function never learns where the file is.

• columns (int) – How many numeric columns the result must have.

• timeout (float) – Seconds before the attempt is abandoned.

Returns:

The rows, as (n, columns) floats.

Return type:

numpy.ndarray

Raises:

ValueError – If the source is rejected, the function raises, it runs too long, or what it returns is not a non-empty table of finite numbers of the right width. Every one of these is a normal outcome, not a bug: the point of checking is that generated code is not trusted to be right.

Examples

>>> code = 'def adapt(text):\n    return [[float(v) for v in line.split(",")]' \
...        '\n            for line in text.splitlines() if line.strip()]'
>>> run_adapter(code, '1,2\n3,4\n', 2).tolist()
[[1.0, 2.0], [3.0, 4.0]]
>>> run_adapter('def adapt(text): return []', '', 2)
Traceback (most recent call last):
ValueError: the generated reader returned no rows

PyHydroGeophysX.agents.runtime.adapters.sample_text(path: Any, lines: int = 12, chars: int = 1200) → str

The first few lines of a text file, for a person and a model to look at.

PyHydroGeophysX.agents.runtime.catalog module

The workflow’s capabilities, registered as tools.

Each entry wraps an existing agent rather than reimplementing it: the agents are the tested part of this codebase and their behaviour must not change here. What changes is who decides to call them, and in what order.

Registration order is dependency order. That is not decoration - it is the fallback policy: without a model to ask, the controller takes the first tool whose requirements are met, which reproduces the pipeline these tools replaced. A run with no API key therefore behaves as it always did.

Each handler returns (summary, outputs). The summary is the sentence the next step and the final report will read, so it states what was found, not that the step finished - “5 surveys, 812 measurements each, 2017-11-05 to 2017-11-09” rather than “loading complete”. This is the shared memory the agents did not previously have: BaseAgent.context is per-agent, so nothing an agent concluded ever reached another one.

PyHydroGeophysX.agents.runtime.catalog.DEFAULT_INSTRUMENT = 'E4D'

The ERT format assumed when the configuration names none. There is no format auto-detect anywhere in this package - the device is chosen, not guessed - so this default has to be readable from outside: the studio panel that opens a run’s data for display must open it the same way the run does, and a second copy of the string here and there would drift.

PyHydroGeophysX.agents.runtime.catalog.MODEL_BUNDLE_DIR = 'ert_model'

Where a run writes its inverted-model bundle, under the run’s output folder. Named here because two things have to agree on it: the inversion step that writes it, and the studio page that opens it while the run is still going.

PyHydroGeophysX.agents.runtime.catalog.PRODUCERS = {'climate': ('fetch_climate',), 'spatial_map': ('map_tdem_plan_view',), 'water_content': ('convert_water_content', 'convert_tdem_water_content', 'convert_mt_water_content')}

The tools that make each product a request can name (_intent.PRODUCTS).

PyHydroGeophysX.agents.runtime.catalog.STUDIO_INVERSION_KEY = 'studio_inversion_dir'

Config key naming the folder the studio wrote a finished ERT inversion to (handover.json beside the model bundle) when a step-by-step task was handed to an automatic run. The run carries on from that model.

PyHydroGeophysX.agents.runtime.catalog.agent_kwargs(ctx: RunContext) → Dict[str, Any]

The constructor arguments every agent takes.

PyHydroGeophysX.agents.runtime.catalog.as_pygimli_data(survey: Any, output_dir: Any, use_source_error: bool = True)

A loaded survey as the pyGIMLi container the inversion agents use.

ERTLoaderAgent returns a StandardERT - the package’s own record, carrying electrodes, observations, instrument and CRS. Several agents instead expect pygimli’s DataContainerERT, which they index (data['k'], data['err']) and query (sensorCount()). Passing the first where the second is expected fails with “‘StandardERT’ object is not subscriptable”, which names neither side of the mismatch.

The conversion is the one the inversion agents already perform: export to a pyGIMLi unified data file, then load it back.

Parameters:

• survey (StandardERT) – A loaded survey. Anything that is already a pyGIMLi container is returned unchanged.

• output_dir (str or Path) – Where the intermediate file is written.

• use_source_error (bool) – Carry the instrument’s own error estimates into the export.

Returns:

The survey in pyGIMLi’s form.

Return type:

DataContainerERT

Raises:

ValueError – If survey is None.

PyHydroGeophysX.agents.runtime.catalog.available_methods(ctx: RunContext) → List[str]

The methods this run actually produced a result for.

Read from the artifacts rather than from config['methods']. The configured list says what the user hoped for; a fusion asked to combine three methods when two succeeded fails with “requires methods: [‘petrophysics’]” instead of combining the two it has.

PyHydroGeophysX.agents.runtime.catalog.build_site_info(ctx: RunContext) → Dict[str, Any]

Site description for the report, from the configuration and file names.

PyHydroGeophysX.agents.runtime.catalog.export_model_bundle(ctx: RunContext, results: Dict[str, Any]) → str

Write the recovered model as the inverted-model bundle, and say where.

The bundle - mesh_res.bms, resmodel.npy, index_marker.npy and optionally all_coverage.npy - is this package’s own interchange format for an inverted model; PyHydroGeophysX.Geophy_modular.ERT_to_WC reads exactly these names, and the studio’s ERT-to-water-content page asks for a folder containing them.

An automatic run used to leave nothing in that format: it held the mesh and the models in memory, converted them in the same process, and wrote only figures. So a run that had just inverted five surveys still left every other part of the studio with nothing to open, and the water-content page asking the user to go and find a model folder that did not exist.

Parameters:

• ctx (RunContext) – The run, for its output directory.

• results (dict) – The inversion agent’s result, carrying mesh and either time_lapse_models or resistivity_model.

Returns:

The folder written, or “” when there was nothing to write or the mesh could not be saved. Exporting is a convenience: a run must not fail because a file could not be written beside the result it already has.

Return type:

str

PyHydroGeophysX.agents.runtime.catalog.fusion_account(ctx: RunContext) → Tuple[List[str], List[str]]

What this run actually combined across methods, and what it did not.

Parameters:

ctx (RunContext) – The run, read for the artifacts the other steps left.

Returns:

(combined, not_combined), one clause per item, for the fusion step’s summary.

Return type:

tuple

Raises:

None –

Examples

>>> ctx = RunContext('x')
>>> ctx.put('seismic_results', {'status': 'success'})
>>> ctx.put('inversion_results', {'status': 'success'})
>>> fusion_account(ctx)
([], ['the seismic model did not constrain the ERT inversion (no structural constraint was derived)'])

PyHydroGeophysX.agents.runtime.catalog.fusion_pattern(methods: Sequence[str]) → str | None

The richest fusion pattern methods can satisfy, or None.

Parameters:

methods (sequence of str) – Methods with a result, as available_methods() reports them.

Returns:

A pattern name, or None when no pattern’s requirements are met.

Return type:

str or None

Raises:

None –

Examples

>>> fusion_pattern(['seismic', 'ert', 'petrophysics'])
'full_integration'
>>> fusion_pattern(['seismic', 'ert'])
'structure_constraint'
>>> fusion_pattern(['ert', 'petrophysics'])
'petrophysics_integration'
>>> fusion_pattern(['ert']) is None
True

PyHydroGeophysX.agents.runtime.catalog.gpr_recipe(profile: Any, config: Mapping[str, Any], velocity: float | None) → Tuple[List[Dict[str, Any]], List[str]]

The processing steps for a run’s GPR line, and a note on each one left out.

The GPR page’s defaults: dewow over three antenna periods, time zero on the first arrival, the mean trace removed and automatic gain control over 25 ns. gpr_params adds a band-pass (bandpass: true for a quarter to 2.5 times the antenna frequency, or [low, high] MHz), migration (migrate, with the run’s velocity), another gain (gain: agc, tpow or none), or replaces the lot (steps, as process() takes them). Either way a migration uses the run’s velocity (gpr_velocity()) and is left out without one.

PyHydroGeophysX.agents.runtime.catalog.gpr_velocity(ctx: RunContext, profile: Any = None) → Tuple[float | None, str]

The radar velocity a run’s depths use, m/ns, and where it came from.

A depth is v t / 2, so it is only as good as the velocity, and a velocity nobody gave must not become one. In order: gpr_velocity (m/ns) or gpr_permittivity from the request, then the permittivity a GSSI file’s header holds - the value set on the instrument when recording, which is said to be that. Without any, (None, reason): the line is reported in two-way time and the run says why there is no depth. A value outside water-to-air (0.01-0.3 m/ns) is not used, and noted.

PyHydroGeophysX.agents.runtime.catalog.interface_coords(seismic: Dict[str, Any], threshold: Any) → Tuple[Any, Any] | None

The (x, z) arrays of one velocity interface, or None.

The two agents disagree about shape, and neither is wrong on its own. SeismicAgent may extract several interfaces at once and returns {threshold: {'x': ..., 'z': ...}}; StructureConstraintAgent builds a mesh from exactly one and unpacks interface_x, interface_z = interface_coords. Handing the dict straight across failed with “not enough values to unpack (expected 2, got 1)” - a message that says nothing about interfaces - so the translation belongs here, named.

Parameters:

• seismic (dict) – A SeismicAgent result.

• threshold (float) – The velocity threshold whose interface is wanted. Falls back to the only interface present when there is exactly one, since a run that extracted one interface meant that one.

Returns:

(x, z), or None when no interface matches.

Return type:

tuple or None

Raises:

None –

Examples

>>> result = {'interfaces': {1200.0: {'x': [0, 1], 'z': [-1, -2]}}}
>>> interface_coords(result, 1200.0)
([0, 1], [-1, -2])
>>> interface_coords(result, 900.0)        # the only one there is
([0, 1], [-1, -2])
>>> interface_coords({'interfaces': {}}, 1200.0) is None
True

PyHydroGeophysX.agents.runtime.catalog.mt_files(config: Mapping[str, Any]) → List[str]

The MT sites the configuration names: transfer-function files or folders of them.

PyHydroGeophysX.agents.runtime.catalog.not_delivered_items(ctx: RunContext, delivered: Mapping[str, Any]) → List[Tuple[str, str]]

(what, why) for everything a report must say it does not contain.

The products the request named and the run did not make, then each step that failed and was not recovered, once and with its reason. A run with a broken TDEM file beside good ERT data used to write a complete-looking report that never mentioned TDEM, and to call itself a success.

Examples

>>> ctx = RunContext('x', {'data_file': 'a.ohm'})
>>> _ = ctx.begin('invert_tdem', description='Run TDEM inversion')
>>> ctx.finish(status='failed', error='ValueError: TDEM data file not found: s.csv')
>>> not_delivered_items(ctx, {})
[('Run TDEM inversion', 'TDEM data file not found: s.csv')]

A step left out is listed with why it was left out:

>>> _ = ctx.begin('load_gravmag_data', description='Load gravity / magnetic data')
>>> ctx.finish(status='skipped', error="Left out with the user's agreement: no gravity data")
>>> not_delivered_items(ctx, {})[-1]
('Load gravity / magnetic data', "Left out with the user's agreement: no gravity data")

PyHydroGeophysX.agents.runtime.catalog.plain_error(error: Any) → str

A step’s error for a reader, without the exception class a tool raised it as.

The tools raise ValueError to explain themselves, and the step records "ValueError: <explanation>"; the class name means nothing to a reader of the report. Any other class is kept, because there it is information.

>>> plain_error("ValueError: no site coordinates")
'no site coordinates'
>>> plain_error("KeyError: 'rhoa'")
"KeyError: 'rhoa'"

PyHydroGeophysX.agents.runtime.catalog.read_gpr_input(path: Any) → Any

The GPR line gpr_file names: a radar file, or the simulated example line.

A folder holding the MODFLOW catchment model of examples/data stands for the example line simulated over it (example_profile()), the line the GPR page opens with “Use example”; the profile then says it is simulated. A folder of radar files is refused with their names, since a run processes one line.

PyHydroGeophysX.agents.runtime.catalog.resolve_path(path: Any, project_dir: Any = '.') → str

A data file’s real location, trying the ways a request gets it wrong.

Requests name files as the user sees them, which is often a bare file name, a path relative to the project directory, or one carrying a duplicated examples/ prefix. This was written out three times in the branch this package replaces, once per file role, with the copies already drifting.

Parameters:

• path (str or Path) – The path as configured.

• project_dir (str or Path, optional) – Directory the request’s files are relative to.

Returns:

The first candidate that exists, or the original path unchanged so the caller reports a missing file by the name the user used.

Return type:

str

Raises:

None –

Examples

>>> resolve_path('absent.dat')
'absent.dat'
>>> resolve_path('')
''
>>> resolve_path(None)
''

PyHydroGeophysX.agents.runtime.catalog.seismic_inversion_params(config: Mapping[str, Any]) → Dict[str, Any]

The seismic inversion settings a configuration gives, under any of its names.

seismic_inversion_params is this runtime’s own key and wins, setting by setting. The request parser writes seismic_params (its fusion stage) and, for a seismic-only request, top-level lam, z_weight and the like; reading the first key alone sent SeismicAgent an empty dictionary, so every value the request stated was replaced by the agent’s defaults.

Examples

>>> seismic_inversion_params({'seismic_params': {'lam': 5, 'zWeight': 1.0}})
{'lam': 5, 'zWeight': 1.0}
>>> seismic_inversion_params({'seismic_params': {'lam': 5},
...                           'seismic_inversion_params': {'lam': 30}, 'z_weight': 0.5})
{'zWeight': 0.5, 'lam': 30}

PyHydroGeophysX.agents.runtime.catalog.shortfall_reasons(ctx: RunContext, results: Mapping[str, Any] | None = None) → Dict[str, str]

Why each product the request can name is missing from this run, where it is.

For unmet_requests(), which on its own can say only that a product is missing; the reason is what the user can act on.

Examples

>>> ctx = RunContext('x', {'user_request': 'estimate the water content'})
>>> _ = ctx.begin('invert_tdem', description='Run TDEM inversion')
>>> ctx.finish(status='failed', error='ValueError: no time column')
>>> shortfall_reasons(ctx)['water_content']
'no resistivity model was recovered to convert: Run TDEM inversion failed (no time column)'

PyHydroGeophysX.agents.runtime.catalog.summarise_run(ctx: RunContext, incomplete: Sequence[str] = ()) → str

The interpretation text, written from the steps that actually ran.

Replaces an f-string that restated the configuration - it reported the configured time-lapse method and the configured regularization whether or not the inversion had used them, because it was written beside the config rather than beside the result.

incomplete - why the run cannot be called complete, when it cannot - heads the text in place of “completed”, so its first line never claims a run finished that did not.

PyHydroGeophysX.agents.runtime.catalog.survey_files(config: Dict[str, Any]) → List[str]

Every ERT survey the configuration names, in order, without duplicates.

PyHydroGeophysX.agents.runtime.context module

The run transcript: what was asked, what was done, and what was found.

One object, passed to every tool, holding two separate things that used to be conflated in a results dictionary:

• Artifacts - the data itself: loaded surveys, recovered models, meshes, arrays. Large, and of interest only to the tool that consumes them.

• Steps - the record of what happened: which tool ran, why, what it concluded, what it could not do. Small, and of interest to everything, including the model choosing the next action and the report describing the run afterwards.

Keeping them apart is what makes a controller loop affordable. The model reads RunContext.transcript(), which is a few hundred words whatever the run did; it never sees a mesh. This is the same discipline a tool-using assistant applies when it summarises a tool result instead of pasting it.

It also removes a whole class of reporting bug. The execution plan is derived from the steps that ran (RunContext.plan()) rather than written by hand beside them, so a plan cannot describe a step that did not happen - which is exactly what it used to do.

PyHydroGeophysX.agents.runtime.context.PROJECTED = <projected>

Stands in for the products of the steps a route has only projected (controller.route_ahead()). A when gate that looks inside an artifact asks RunContext.projected() first and answers from the configuration instead, since there is nothing inside yet.

class PyHydroGeophysX.agents.runtime.context.RunContext(goal: str = '', config: Dict[str, Any] | None = None, output_dir: str = '', settings: Dict[str, Any] | None = None)

Bases: object

Everything one workflow run knows about itself.

Parameters:

• goal (str) – The user’s request, in their own words.

• config (dict, optional) – The parsed workflow configuration.

• output_dir (str, optional) – Where artifacts are written.

• settings (dict, optional) – Anything a tool needs that is not configuration - API key, model name, provider, progress callback.

Examples

>>> ctx = RunContext('invert this line', {'ert_file': 'a.dat'})
>>> ctx.put('ert_data', object(), 'Loaded 1 survey, 812 measurements.')
>>> ctx.has('ert_data')
True
>>> ctx.begin('invert_ert', 'the data are loaded').tool
'invert_ert'
>>> ctx.finish(summary='Converged to chi-squared 1.6.')
>>> ctx.steps[-1].status
'ok'

artifacts: Dict[str, Any]

ask(question: str, options: List[Dict[str, str]], default: str = '') → str

Put a choice to the user and wait, or take default if nobody is there.

For the cases a run genuinely cannot decide for itself: two files that disagree about their origin, a conversion with no calibration, a product that can be produced two defensible ways. Guessing there is worse than asking, and failing is worse than either - the run has usually done a lot of work by the time it finds out.

Parameters:

• question (str) – What is being asked, in one or two sentences, stating the evidence.

• options (list of dict) – {"id", "label", "detail"} per choice. id is what comes back; detail says what that choice will actually do.

• default (str) – The id to use when there is no one to ask - a headless run, a script, a test. Defaults to the first option.

Returns:

The chosen option’s id. Always one of the offered ids: an answer that matches nothing falls back to default, because a typo must not select behaviour nobody offered.

Return type:

str

Raises:

None –

Examples

>>> ctx = RunContext('goal')
>>> options = [{'id': 'a', 'label': 'Use the station file'},
...            {'id': 'b', 'label': 'Use the SEG-Y headers'}]
>>> ctx.ask('Which origin is right?', options)   # nobody to ask
'a'
>>> ctx.warnings[0].startswith('Nobody was available')
True
>>> ctx.settings['ask_user'] = lambda event: 'b'
>>> ctx.ask('Which origin is right?', options)
'b'

attempted(tool: str) → int

How many times tool has been tried, successfully or not.

A blocked step is not an attempt: the tool never ran, because something it needed was missing or it was not on offer. Counting it made a premature choice permanent - a conversion named before the evaluation existed was never offered again once the evaluation did exist.

begin(tool: str, reason: str = '', agent: str = '', description: str = '', purpose: str = '') → Step

Open a step. Only one is open at a time.

config: Dict[str, Any]

descriptions: Dict[str, str]

A one-line description of each artifact, for the transcript. The artifact itself may be a hundred megabytes; this is what the model gets to see about it.

elapsed() → float

ended

finished, exhausted, stopped or step_limit (see controller), “” until it has run. A run the user or the step limit cut off did not run out of work, and whether it is complete depends on which it was.

Type:

How the controller’s loop ended

finish(summary: str = '', status: str = 'ok', error: str = '') → None

Close the open step with what it concluded.

get(key: str, default: Any = None) → Any

guidance: List[str]

What the user told the run while it worked (steering), in order. Part of the transcript, so every later decision is made with it in view, not only the one that first read it.

has(*keys: str) → bool

True when every named artifact is present and not None.

note(message: str) → None

A warning the user should see, deduplicated.

plan() → List[Dict[str, Any]]

The execution plan, derived from what actually ran.

Returns the shape the report and the audit already consume, so a plan can no longer disagree with the run it describes.

projected(key: str) → bool

True when key exists only in a projected route, not yet in fact.

>>> ctx = RunContext('x')
>>> ctx.put('data', PROJECTED)
>>> ctx.has('data'), ctx.projected('data')
(True, True)

proposals: List[Dict[str, Any]]

what was proposed and whether it was accepted, in order.

Type:

Off-route actions put to the user (toolbox)

put(key: str, value: Any, description: str = '') → None

Record an artifact and how to describe it.

questions: List[Dict[str, Any]]

Choices the run put to the user and the answers it got. Recorded so the report can say a result rests on a decision somebody made, which is otherwise invisible in the numbers.

ran(tool: str) → bool

True when tool has already completed successfully.

reopen(names: Iterable[str]) → None

Let names run again: their earlier attempts no longer count.

For the steps downstream of one that was run again with other settings: what they read has changed, so what they made is out of date.

Examples

>>> ctx = RunContext('x')
>>> _ = ctx.begin('evaluate'); ctx.finish()
>>> ctx.attempted('evaluate')
1
>>> ctx.reopen(['evaluate'])
>>> ctx.attempted('evaluate')
0

reopened: Dict[str, int]

Tool name -> index of the first step that counts as an attempt of it. A step run again with other settings changes what the steps after it read; those are reopened (reopen()), so they can run again on the new result instead of being refused as already tried.

results() → Dict[str, Any]

The artifacts, as the dictionary the existing callers expect.

settings: Dict[str, Any]

steps: List[Step]

transcript(limit: int = 40) → str

The run so far, as the controller reads it.

Parameters:

limit (int) – Most recent steps to include. A run that loops is the one that most needs its recent history; the early steps are represented by the artifacts they produced.

Returns:

Goal, artifacts on hand, and the step log.

Return type:

str

unfinished() → List[str]

Tools that were attempted and did not succeed.

warnings: List[str]

PyHydroGeophysX.agents.runtime.context.STATUSES = ('ok', 'failed', 'blocked', 'skipped')

the tool never ran because something it needed was absent, which is a fact about the run rather than an error in the tool.

Type:

A step’s outcome. blocked is distinct from failed

class PyHydroGeophysX.agents.runtime.context.Step(tool: str, reason: str = '', status: str = 'ok', summary: str = '', error: str = '', produced: List[str] = <factory>, seconds: float = 0.0, agent: str = '', description: str = '', purpose: str = '')

Bases: object

One tool invocation and what came of it.

agent: str = ''

description: str = ''

error: str = ''

line() → str

The step as one line of transcript.

produced: List[str]

purpose: str = ''

What the tool is for, in its own words. Distinct from reason, which is why this run chose it; without a model to ask there is no reason, and the plan then read “next available step” for every row.

reason: str = ''

Why this step was chosen, in the controller’s words. Recorded because a reader of the finished report asks “why did it do that”, and the answer is otherwise nowhere.

seconds: float = 0.0

status: str = 'ok'

summary: str = ''

One or two sentences of what the tool concluded - not its output data.

tool: str

PyHydroGeophysX.agents.runtime.context.merge_outputs(ctx: RunContext, outputs: Dict[str, Any], describe: Dict[str, str] | None = None, keys: Iterable[str] | None = None) → List[str]

Store a tool’s outputs as artifacts, skipping empties.

Parameters:

• ctx (RunContext) – The run to store into.

• outputs (dict) – What the tool returned.

• describe (dict, optional) – Per-key one-line descriptions for the transcript.

• keys (iterable of str, optional) – Restrict to these keys; by default everything that is not None and not bookkeeping (status, error) is stored.

Returns:

The keys that were stored.

Return type:

list of str

Examples

>>> ctx = RunContext('x')
>>> merge_outputs(ctx, {'status': 'success', 'model': [1, 2], 'empty': None})
['model']

PyHydroGeophysX.agents.runtime.controller module

The loop: read the transcript, choose one action, observe, choose again.

This is the whole difference from the pipeline it replaces. A pipeline decides everything before it starts; a loop decides one step at a time, with the results of the previous steps in front of it. That is what lets a run react - to an inversion that did not converge, to a mesh with no layering, to a file that turned out to hold something other than what its name suggested.

Two properties matter more than the loop itself:

It always terminates, and always produces something. A model that keeps choosing badly is bounded by max_steps; a model that cannot be reached at all falls through to next_by_policy(), which walks the tools in dependency order. A run without an API key therefore behaves exactly like the old pipeline - it simply stops adapting. This matters because the previous design’s failure mode was an exception mid-sequence that lost the whole run.

A refusal is data. When the controller names a tool that cannot run, or one that does not exist, that becomes a recorded step with the reason, and the loop continues. The model sees its own mistake in the transcript on the next turn, which is how it corrects. Raising instead would end the run over a typo.

PyHydroGeophysX.agents.runtime.controller.BROKE = 'broke'

The loop’s own code failed again after the run had resumed past it (entry.drive()).

PyHydroGeophysX.agents.runtime.controller.DEFAULT_FINISH = 'report_files'

The artifact that has to exist before the controller may finish. AQUAH’s runs end with a report; an assistant whose last product is something else - a reviewed report, say - passes its own key as finish.

PyHydroGeophysX.agents.runtime.controller.DEFAULT_MAX_STEPS = 24

A run that has taken this many steps is looping rather than progressing. Generous: a time-lapse run with a retry and a report is about eight.

PyHydroGeophysX.agents.runtime.controller.EXHAUSTED = 'exhausted'

the model finished once a report existed, nothing more could run, the user stopped the run, or it used every step it was allowed.

Type:

How the loop ended, as RunContext.ended records it

PyHydroGeophysX.agents.runtime.controller.FINISHED = 'finished'

the model finished once a report existed, nothing more could run, the user stopped the run, or it used every step it was allowed.

Type:

How the loop ended, as RunContext.ended records it

PyHydroGeophysX.agents.runtime.controller.FLEXIBLE_MAX_STEPS = 40

the standard route is about eight, and the rest is room for looks, a re-run and the steps after it.

Type:

Steps a flexible run may take

PyHydroGeophysX.agents.runtime.controller.PROCEED = 'proceed'

What an on_step hook may answer. proceed is the default for any other value, including None, so a hook that forgets to return cannot stall a run.

PyHydroGeophysX.agents.runtime.controller.SKIP = 'skip'

What an on_step hook may answer. proceed is the default for any other value, including None, so a hook that forgets to return cannot stall a run.

PyHydroGeophysX.agents.runtime.controller.STEERING_PROMPT = '\nThe user has just told you, while the run is going:\n{notes}\nFollow it where you can. If it asks for a different setting, add the changed\nkeys to your JSON as "settings", using these current values as the template\n(change only what the user asked for):\n{settings}\nIf it asks for something none of the actions or settings can do, say so in\n"why" and carry on with the best next action.\n'

Added to the decision prompt when the user has just said something. The settings it may change are the recovery’s (ADJUSTABLE): numbers and switches between behaviours already implemented, never a path.

PyHydroGeophysX.agents.runtime.controller.STEP_LIMIT = 'step_limit'

the model finished once a report existed, nothing more could run, the user stopped the run, or it used every step it was allowed.

Type:

How the loop ended, as RunContext.ended records it

PyHydroGeophysX.agents.runtime.controller.STOP = 'stop'

What an on_step hook may answer. proceed is the default for any other value, including None, so a hook that forgets to return cannot stall a run.

PyHydroGeophysX.agents.runtime.controller.STOPPED = 'stopped'

the model finished once a report existed, nothing more could run, the user stopped the run, or it used every step it was allowed.

Type:

How the loop ended, as RunContext.ended records it

PyHydroGeophysX.agents.runtime.controller.forced_choice(ctx: RunContext, tools: Dict[str, Any] | None = None, finish: str = 'report_files') → Dict[str, Any] | None

The decision to take without asking, when the menu leaves only one.

The model may answer with a tool from the menu, or with “done” once a report has been written (CONTROLLER_PROMPT). With one tool on the menu and no report yet there is exactly one valid answer, and asking for it cost a full model request: three of the six decisions in an ERT-to-water- content run - load, invert, write the report - were of this kind. Taking it directly is the same decision, and the step says so, so the plan and the transcript show that nothing was chosen.

Parameters:

• ctx (RunContext) – The run as it stands.

• tools (dict, optional) – The registry to choose from; the global one by default.

• finish (str) – The artifact whose existence makes “done” a valid answer.

Returns:

{"tool": name, "why": "only one action was available: name"} when the choice is forced; None when the model has something to decide, including whether to finish.

Return type:

dict or None

Raises:

None –

Examples

>>> from .tools import Tool
>>> only = {'load': Tool('load', 'Load.', lambda c: ('', {'data': 1}),
...                      produces=('data',))}
>>> forced_choice(RunContext('x'), only)
{'tool': 'load', 'why': 'only one action was available: load'}
>>> ctx = RunContext('x')
>>> ctx.put('report_files', {'report_markdown': 'r.md'})
>>> forced_choice(ctx, only) is None        # "done" is a valid answer now
True

PyHydroGeophysX.agents.runtime.controller.next_by_policy(ctx: RunContext, tools: Dict[str, Any] | None = None) → str | None

The next tool when there is no model to ask.

Takes the first tool whose requirements are met and which has not already run. Registration order is dependency order, so this reproduces the fixed pipeline - which is the right fallback: without a model the run should still do the obvious thing, it just cannot reconsider.

Parameters:

ctx (RunContext) – The run as it stands.

Returns:

A tool name, or None when nothing can run.

Return type:

str or None

Raises:

None –

Examples

>>> from .tools import Tool
>>> ctx = RunContext('x')
>>> only = {'first': Tool('first', 'One.', lambda c: ('', {'a': 1}),
...                       produces=('a',)),
...         'second': Tool('second', 'Two.', lambda c: ('', {}),
...                        requires=('a',))}
>>> next_by_policy(ctx, only)
'first'
>>> ctx.put('a', 1)
>>> _ = ctx.begin('first'); ctx.finish()
>>> next_by_policy(ctx, only)
'second'

PyHydroGeophysX.agents.runtime.controller.route_ahead(ctx: RunContext, tools: Dict[str, Any] | None = None, finish: str = 'report_files', limit: int = 12) → List[Dict[str, str]]

The steps still between the run as it stands and its final product.

Worked out on a copy of the run: next_by_policy() takes a step, the step is assumed to make what it declares, and so on until finish exists or nothing more can run. It is the dependency order applied to what the run actually has, not a promise - the controller may choose differently, a step may fail - which is why the desktop recomputes it after every step and calls it the route as things stand. The products of a projected step are PROJECTED, which the when gates that look inside an artifact recognise.

Parameters:

• ctx (RunContext) – The run as it stands; not modified.

• tools (dict, optional) – The registry to project over; the global one by default.

• finish (str) – The artifact that ends the run.

• limit (int) – At most this many steps are projected.

Returns:

{"tool", "label", "module"} per step, in order. Empty once the final product exists, or when nothing can run.

Return type:

list of dict

Raises:

None –

Examples

>>> from .tools import Tool
>>> chain = {'load': Tool('load', 'Load.', lambda c: ('', {}), produces=('data',),
...                       label='Load data'),
...          'report': Tool('report', 'Report.', lambda c: ('', {}),
...                         requires=('data',), produces=('report_files',),
...                         label='Write the report')}
>>> [step['label'] for step in route_ahead(RunContext('x'), chain)]
['Load data', 'Write the report']
>>> ctx = RunContext('x')
>>> ctx.put('data', [1])
>>> _ = ctx.begin('load'); ctx.finish()
>>> [step['tool'] for step in route_ahead(ctx, chain)]
['report']

PyHydroGeophysX.agents.runtime.controller.run_controller(ctx: RunContext, ask: Callable[[str], str] | None = None, max_steps: int = 24, progress: Callable[[str, float, str], None] | None = None, tools: Dict[str, Any] | None = None, on_step: Callable[[...], str | None] | None = None, on_result: Callable[[...], None] | None = None, recovery: bool = True, prompt: str | None = None, finish: str = 'report_files', flexible: bool = False, approves_steps: bool = False) → RunContext

Drive the run to completion, one chosen action at a time.

Parameters:

• ctx (RunContext) – The run. Mutated in place and returned.

• ask (callable, optional) – Takes a prompt, returns the model’s reply. Omit to run the deterministic policy - which is what happens without an API key.

• max_steps (int) – Hard bound on iterations, so a model that keeps choosing badly still terminates with whatever it has produced.

• progress (callable, optional) – (step, fraction, detail), called before each action so the desktop app can show the run advancing.

• on_step (callable, optional) – (ctx, tool, reason) -> 'proceed' | 'skip' | 'stop', called before each action with the tool about to run. This is the single difference between the app’s two modes: “auto to report” supplies a hook that notes the tool’s module and returns immediately, while “step by step” supplies one that shows the user what is about to happen and waits. Both drive the same loop over the same tools, so a step behaves identically whichever way it was reached - the previous design had two separate code paths that could and did drift.

• recovery (bool) – Whether a failed step may be diagnosed and retried once with different settings. On by default, because the alternative - recording the failure and moving on - loses runs that one changed number would have saved. Only configuration is ever changed, never code and never a file path; see recovery.

• on_result (callable, optional) – (ctx, step), called after each action with the step it finished. on_step fires before a step and so can only say what is about to happen; this is where what a step actually concluded becomes available to a caller while the run is still going.

• prompt (str, optional) – The decision prompt, with {transcript} and {menu} fields. Defaults to CONTROLLER_PROMPT; an assistant for another domain passes its own framing.

• finish (str) – The artifact that must exist before “done” is accepted (DEFAULT_FINISH, the report, unless the assistant says otherwise).

• flexible (bool) – Let the model step off the standard route (toolbox): look at a product’s numbers, add a step the route has not scheduled, run a finished step again with other settings, change a setting, leave route steps out, write a temporary script, or ask the user. Everything but a look is put to the user first with the model’s reason (toolbox.propose()), and the route is the default. The model is asked with FLEXIBLE_PROMPT, and says in its answer which steps it means to run next and which results it wants to see; those steps then run without asking it again (_follow_plan()) until it wants to look, or something it could not have planned for happens. Ignored without ask, and with a prompt of the assistant’s own, which does not offer it.

• approves_steps (bool) – on_step asks the user before every step (step by step), so an off-route step is approved there, with its reason, rather than in a question of its own.

Returns:

The same context, carrying the steps taken, the artifacts produced, any warnings raised and, in ended, how the loop ended. A model’s “done” ends it only once a report exists; before that the next available step is taken instead.

Return type:

RunContext

Raises:

None – Tool failures are recorded, not raised: the point of the loop is to keep deciding after one.

Examples

>>> from .tools import Tool
>>> only = {'load': Tool('load', 'Load.',
...                      lambda c: ('Loaded 3 files.', {'data': [1, 2, 3]}),
...                      produces=('data',)),
...         'report': Tool('report', 'Report.', lambda c: ('Wrote it.', {}),
...                        requires=('data',))}
>>> ctx = run_controller(RunContext('process my data'), tools=only)
>>> [s.tool for s in ctx.steps]
['load', 'report']
>>> ctx.steps[0].summary
'Loaded 3 files.'
>>> ctx.ended
'exhausted'

PyHydroGeophysX.agents.runtime.entry module

The workflow entry point, as a controller run.

BaseAgent.run_unified_agent_workflow keeps its signature and its four-value return, because the desktop app, the Streamlit app and the one-click runner all call it. What changed is everything behind that signature: instead of classifying the request into one of eight workflow types and running that type’s fixed sequence, it builds a run context and lets the controller choose each step from the tools whose inputs exist.

The old implementation is still present as BaseAgent.run_legacy_agent_workflow. It is not dead weight and not indecision: it covers workflow types whose tools cannot be exercised on this machine - there is no ParFlow output, no SEG-Y file and no GPU here - and keeping it reachable means a run that behaves differently can be compared against the code it replaced instead of argued about. Set PHGX_LEGACY_WORKFLOW=1 to use it.

PyHydroGeophysX.agents.runtime.entry.LEGACY_ENV = 'PHGX_LEGACY_WORKFLOW'

Set to 1 to run the pre-controller implementation instead.

PyHydroGeophysX.agents.runtime.entry.MAX_LOOP_ERRORS = 2

How many times drive() picks a run up after an error in the loop’s own code before it stops choosing steps.

PyHydroGeophysX.agents.runtime.entry.adopt_request_inputs(config: Dict[str, Any]) → Tuple[Dict[str, Any], List[str]]

The configuration with what its request states and it leaves out, and a note on each.

A configuration of a request alone - {"user_request": "invert a.ohm and estimate water content"} - used to end with “No step could run with the inputs given”: nothing read the files the request names unless the model parser had run first. The rules are those of AgentCoordinator._resolve_config: only what the request itself states is taken (ContextInputAgent.request_inputs(), never the parser’s defaults), the ERT files only when the configuration names no ERT data, and nothing the caller set is replaced - except that a file another method takes is not also ERT data (drop_borrowed_ert_files()).

Parameters:

config (dict) – The workflow configuration, with user_request.

Returns:

(config, notes) - a new dictionary, and one sentence per value taken from the request or named there and not used.

Return type:

tuple

Raises:

None –

Examples

>>> config, notes = adopt_request_inputs({'user_request': 'invert a.ohm with a Syscal'})
>>> config['data_file'], config['instrument']
('a.ohm', 'Syscal')
>>> print(notes[0])
The configuration names no ERT data, so the run uses the file the request names: a.ohm.
>>> adopt_request_inputs({'data_file': 'b.ohm', 'user_request': 'invert b.ohm'})[1]
[]
>>> adopt_request_inputs({'data_file': 'p.tiw', 'tdem_file': 'p.tiw'})[0]
{'tdem_file': 'p.tiw'}

Only the user’s words count, and not data for a method the model ruled out:

>>> request = ('Invert a.ohm and estimate water content.\n\nAttached local files (use '
...            'only for their listed roles):\nchat_reference: notes.txt\n\nTDEM '
...            'soundings are read from sounding.csv.')
>>> 'tdem_file' in adopt_request_inputs({'user_request': request})[0]
False
>>> read = {'user_request': 'Invert a.ohm; the TDEM file s.csv is from another site.',
...         'request_read_by': 'model', 'request_methods': {'tdem': False}}
>>> 'tdem_file' in adopt_request_inputs(read)[0]
False

PyHydroGeophysX.agents.runtime.entry.announced(on_step: Callable[[...], str | None], announce: Callable[[Dict[str, Any]], None]) → Callable[[...], str | None]

on_step, followed by auto mode’s “now running” event once it approves.

Step-by-step answered the approval and then said nothing until the step had finished, so the desktop showed “Continuing · proceed” for the whole of an inversion and the choice of the next step after it. Auto mode announces each step as it starts; this sends the same announcement once the user has approved one, and nothing for a step that was skipped or that ended the run.

Parameters:

• on_step (callable) – The pause hook, as run_controller() takes it.

• announce (callable) – Receives the step’s announcement().

Returns:

A hook returning exactly what on_step returned. An announce that raises is ignored, as auto mode ignores it: a display must not stop a run.

Return type:

callable

Raises:

None – What on_step raises propagates, for the controller to handle as it always has.

Examples

>>> from .tools import Tool
>>> seen = []
>>> hook = announced(lambda ctx, tool, why: 'proceed', seen.append)
>>> tool = Tool('invert_ert', 'Invert.', lambda c: ('', {}), label='Run ERT inversion')
>>> hook(RunContext('goal'), tool, 'the data are loaded')
'proceed'
>>> seen[0]['label']
'Run ERT inversion'
>>> announced(lambda ctx, tool, why: 'skip', seen.append)(RunContext('g'), tool, '')
'skip'
>>> len(seen)
1

PyHydroGeophysX.agents.runtime.entry.drive(ctx: RunContext, *, tools: Dict[str, Any] | None = None, ask: Callable[[str], str] | None = None, progress_callback: Callable[[...], None] | None = None, on_step: Callable[[...], str | None] | None = None, on_event: Callable[[Dict[str, Any]], None] | None = None, prompt: str | None = None, finish: str = 'report_files', recovery: bool = True, max_steps: int = 24, flexible: bool = False) → RunContext

Run the controller over tools, telling the caller about every step.

This is the agent loop any assistant runs on (see PyHydroGeophysX.agents.assistants): the controller picks each step from the tools whose inputs exist, and every step is announced before it runs and reported after it ends - through progress_callback for a progress line, and through on_event as structured events, which is what the studio’s live timeline, approvals and chat narration are built on. An assistant that drives its tools through this gets all of that for nothing.

Parameters:

• ctx (RunContext) – The run: goal, configuration, output folder and settings. Mutated in place and returned.

• tools (dict, optional) – The assistant’s tool registry, name to Tool. Defaults to the global registry, which holds AQUAH’s tools.

• ask (callable, optional) – Takes a prompt, returns the model’s reply. Without it the tools run in dependency order.

• progress_callback (callable, optional) – (step, fraction, detail, module); the fourth argument is optional.

• on_step (callable, optional) – The pause policy, as for run_workflow().

• on_event (callable, optional) – Receives phase="start" and phase="done" events for every step, and a phase="route" event - ahead, the steps still between the run and finish as things stand (route_ahead()) - before the first step and after each one. While the model decides, phase="thought" events carry its reasoning as it writes it, when ask takes on_text (as make_ask()’s does).

• prompt – Passed to run_controller(): the decision prompt, the artifact that has to exist before the run may finish, whether a failed step may be retried with changed settings, the step limit, and whether the model may step off the standard route with the user’s agreement.

• finish – Passed to run_controller(): the decision prompt, the artifact that has to exist before the run may finish, whether a failed step may be retried with changed settings, the step limit, and whether the model may step off the standard route with the user’s agreement.

• recovery – Passed to run_controller(): the decision prompt, the artifact that has to exist before the run may finish, whether a failed step may be retried with changed settings, the step limit, and whether the model may step off the standard route with the user’s agreement.

• max_steps – Passed to run_controller(): the decision prompt, the artifact that has to exist before the run may finish, whether a failed step may be retried with changed settings, the step limit, and whether the model may step off the standard route with the user’s agreement.

• flexible – Passed to run_controller(): the decision prompt, the artifact that has to exist before the run may finish, whether a failed step may be retried with changed settings, the step limit, and whether the model may step off the standard route with the user’s agreement.

Returns:

ctx, with the steps taken and what they produced.

Return type:

RunContext

Raises:

None – Tool failures are recorded as steps; display callbacks that raise are ignored; an error in the controller’s own code becomes a note and the run resumes (MAX_LOOP_ERRORS).

Examples

>>> from .tools import Tool
>>> tools = {'load': Tool('load', 'Load.', lambda c: ('Loaded.', {'data': 1}),
...                       produces=('data',)),
...          'summarise': Tool('summarise', 'Summarise.',
...                            lambda c: ('Done.', {'summary': 1}),
...                            requires=('data',), produces=('summary',))}
>>> seen = []
>>> ctx = drive(RunContext('goal'), tools=tools, finish='summary',
...             on_event=seen.append)
>>> [s.tool for s in ctx.steps]
['load', 'summarise']
>>> [(e['phase'], e.get('tool', '')) for e in seen[:4]]
[('route', ''), ('start', 'load'), ('done', 'load'), ('route', '')]
>>> [step['tool'] for step in seen[0]['ahead']], seen[3]['ahead'][0]['tool']
(['load', 'summarise'], 'summarise')

PyHydroGeophysX.agents.runtime.entry.make_ask(api_key: str | None, model: str | None, provider: str) → Callable[[str], str] | None

Public name for assistants building their own workflow on drive().

PyHydroGeophysX.agents.runtime.entry.run_workflow(workflow_config: Dict[str, Any], api_key: str | None, llm_model: str | None, llm_provider: str, output_dir: Any, progress_callback: Callable[[str, float, str], None] | None = None, on_step: Callable[[...], str | None] | None = None, ask_user: Callable[[Dict[str, Any]], str] | None = None, on_event: Callable[[Dict[str, Any]], None] | None = None) → Tuple[Dict[str, Any], list, str, Dict[str, str]]

Run one workflow and return what its callers expect.

Parameters:

• workflow_config (dict) – The parsed configuration, as ContextInputAgent produces it.

• api_key (str) – Model access. Without a key the run still executes, deterministically.

• llm_model (str) – Model access. Without a key the run still executes, deterministically.

• llm_provider (str) – Model access. Without a key the run still executes, deterministically.

• output_dir (str or Path) – Where results are written.

• progress_callback (callable, optional) – (step, fraction, detail, module) for the desktop progress bar; the fourth argument is optional and older callbacks still work.

• on_step (callable, optional) – The pause policy. Omit for “auto to report”; pass step_by_step() to walk the run one approved step at a time.

• ask_user (callable, optional) – Answers a question a tool could not decide for itself, given {"question", "options", "default"} and returning an option id. Without it a run takes the stated default and records that nobody was asked, so a headless run never hangs waiting for an answer.

• on_event (callable, optional) – Receives each step as a structured event: the announcement() with phase="start" before it runs, and its completion() with phase="done" after. The progress callback flattens both into one line of text; this keeps the controller’s reason, the step’s status and what it produced apart, which is what the studio’s live timeline shows.

Returns:

(results, execution_plan, interpretation, report_files) - the same four values the pipeline returned. execution_plan is now derived from the steps that ran, so it can no longer describe a step that did not happen. results["status"] is "incomplete" when the run cannot be called complete (_incomplete_because()), with the reasons among results["warnings"] and heading the interpretation.

Return type:

tuple

Raises:

None – Tool failures are recorded and reported rather than raised, so a run that got most of the way still delivers what it has; one where no step succeeded comes back "incomplete" with every failure in its warnings.

Examples

>>> results, plan, text, files = run_workflow(
...     {'user_request': 'nothing to do'}, None, None, 'openai', '.')
>>> plan, results['status']
([], 'incomplete')
>>> text.splitlines()[1]
'- No step could run with the inputs given, so the run produced nothing.'

PyHydroGeophysX.agents.runtime.entry.use_legacy() → bool

Whether this process should run the pre-controller implementation.

PyHydroGeophysX.agents.runtime.modes module

Auto and step-by-step, as two hooks on one loop.

The studio offers the same work two ways: “Auto to report” runs to the end, and the chat walks through it a step at a time. Those were separate code paths - different dispatch, different ordering, different error handling - so a step could behave one way when a user approved it and another when the workflow ran it unattended, and a fix to one did not reach the other.

They are now the same run_controller() over the same registry, differing only in the on_step hook they supply:

• auto() never pauses. It announces the step and where in the studio it is happening, so the app can bring that module to the front while the work runs.

• step_by_step() asks first and honours the answer, including “skip this one” and “stop here”.

Both announce the same thing in the same order, which is what makes the two modes feel like one feature rather than two implementations.

Nothing here imports Qt. A hook reports a module key; what the desktop does with it - navigate, highlight, ignore - is the desktop’s decision, and the Streamlit app and the tests can pass their own.

PyHydroGeophysX.agents.runtime.modes.announcement(ctx: RunContext, tool: Any, reason: str) → Dict[str, Any]

What the app is told about the step that is about to run.

Parameters:

• ctx (RunContext) – The run as it stands.

• tool (Tool or None) – The tool about to run. None when the controller named something that is not registered, which is itself worth announcing.

• reason (str) – Why the controller chose it.

Returns:

{"tool", "label", "module", "reason", "step"} - all JSON-safe. module is a studio module key or "".

Return type:

dict

Raises:

None –

Examples

>>> from .tools import Tool
>>> ctx = RunContext('goal')
>>> tool = Tool('invert_ert', 'Invert.', lambda c: ('', {}),
...             label='Run ERT inversion', module='ert')
>>> announcement(ctx, tool, 'the data are loaded') == {
...     'tool': 'invert_ert', 'label': 'Run ERT inversion', 'module': 'ert',
...     'reason': 'the data are loaded', 'step': 1}
True
>>> announcement(ctx, None, '')['label']
'Unknown step'

PyHydroGeophysX.agents.runtime.modes.auto(announce: Callable[[Dict[str, Any]], None] | None = None) → Callable[[...], str]

An on_step hook that runs straight through, narrating as it goes.

Parameters:

announce (callable, optional) – Receives the announcement() for each step before it runs. The desktop uses it to bring the relevant module to the front, so the user watches the work happen rather than a progress bar.

Returns:

Suitable as run_controller(..., on_step=...). Always proceeds; an announcement that raises is ignored, because a display problem must not stop a computation.

Return type:

callable

Raises:

None –

Examples

>>> from .tools import Tool
>>> seen = []
>>> hook = auto(seen.append)
>>> tool = Tool('invert_ert', 'Invert.', lambda c: ('', {}), module='ert')
>>> hook(RunContext('goal'), tool, 'because')
'proceed'
>>> seen[0]['module']
'ert'
>>> auto(lambda event: 1 / 0)(RunContext('g'), tool, '')
'proceed'

PyHydroGeophysX.agents.runtime.modes.collect(events: List[Dict[str, Any]]) → Callable[[Dict[str, Any]], None]

An announce that appends to a list, for tests and for replay.

Examples

>>> seen = []
>>> collect(seen)({'tool': 'x'})
>>> seen
[{'tool': 'x'}]

PyHydroGeophysX.agents.runtime.modes.completion(ctx: RunContext, step: Any, tools: Dict[str, Any] | None = None) → Dict[str, Any]

What the app is told about a step that has just finished.

The counterpart to announcement(). That one fires before a step and so can only say what is about to happen; this carries what the step actually concluded, which is the only evidence of work a user can see while a run is still going - the studio’s own panels stay empty, because the workflow runs headless in its own process.

Returns:

{"tool", "label", "module", "summary", "status", "produced", "step"} - all JSON-safe, since this crosses a process boundary as one line.

Return type:

dict

Examples

>>> ctx = RunContext('goal')
>>> ctx.begin('load_ert_surveys', 'it is first',
...           description='Load ERT data')
...
Step(tool='load_ert_surveys', ...)
>>> ctx.put('ert_data', [1, 2])
>>> ctx.finish('Loaded 5 surveys, 812 measurements.')
>>> event = completion(ctx, ctx.steps[-1])
>>> event['summary'], event['status'], event['produced']
('Loaded 5 surveys, 812 measurements.', 'ok', ['ert_data'])

PyHydroGeophysX.agents.runtime.modes.step_by_step(approve: Callable[[Dict[str, Any]], Any], announce: Callable[[Dict[str, Any]], None] | None = None) → Callable[[...], str]

An on_step hook that asks before each step and honours the answer.

Parameters:

• approve (callable) – Receives the announcement() and returns what to do: True or 'proceed' to run it, 'skip' to leave it out, 'stop' or False to end the run here. Blocking is expected - in the desktop this is where the user reads the step and presses a button.

• announce (callable, optional) – Called before approve, with the same event, so the app can navigate to the module first and let the user see the panel they are deciding about.

Returns:

Suitable as run_controller(..., on_step=...).

Return type:

callable

Raises:

None – An approve that raises is treated as “stop”: a run whose approval gate is broken must not carry on unapproved.

Examples

>>> from .tools import Tool
>>> tool = Tool('invert_ert', 'Invert.', lambda c: ('', {}), module='ert')
>>> ctx = RunContext('goal')
>>> step_by_step(lambda event: True)(ctx, tool, '')
'proceed'
>>> step_by_step(lambda event: 'skip')(ctx, tool, '')
'skip'
>>> step_by_step(lambda event: False)(ctx, tool, '')
'stop'
>>> step_by_step(lambda event: 1 / 0)(ctx, tool, '')
'stop'

PyHydroGeophysX.agents.runtime.recovery module

What the loop does about a step that failed.

Before this, a failed step was recorded and the run moved on. That is right for a step nothing can be done about, and wrong for the ones that fail because a setting was unsuitable rather than because the work is impossible - an inversion that diverged at the configured regularisation, a conversion given a threshold no cell meets, a loader pointed at the wrong column layout. A person looking at the error would change one number and run it again, and the loop could not.

The inversion already had a hand-written version of this: evaluate the result, adjust lambda, invert again. That special case is what this generalises.

Three things are deliberately separated:

• Diagnosis is the model’s, because reading an error message and saying what it means is what a model is for. It never runs anything.

• The decision is bounded here: the only recovery this performs on its own is re-running the same tool with different configuration. Changing numbers is reversible, inspectable, and recorded in the plan; it cannot reach outside the run.

• Anything beyond configuration - a generated reader for a coordinate file nobody anticipated - is a proposal that goes to the user through RunContext.ask() with the source in front of them, and runs only on an explicit yes. See adapters. Code a model wrote, executed unattended against somebody’s data, is not a thing this package does on its own initiative: a run with nobody to ask declines.

A recovery is itself a step in the transcript, with the diagnosis as its reason, so a reader of the report sees what failed, what the run thought about it, what it changed, and whether that worked.

PyHydroGeophysX.agents.runtime.recovery.ADJUSTABLE = ('inversion_params', 'temporal_regularization', 'time_lapse_method', 'quality_threshold', 'max_attempts', 'auto_adjust', 'velocity_threshold', 'seismic_inversion_params', 'align_origin', 'petrophysical_params', 'n_realizations', 'coverage_threshold', 'instrument', 'crs', 'first_break_params', 'tdem_params', 'mt_params', 'gravmag_params', 'gravmag_kind', 'magnetic_field', 'gpr_params')

Configuration keys a diagnosis is allowed to change. Everything here is a number or a switch that selects between behaviours already implemented and tested. Absent on purpose: every path key, because a recovery that can point the run at a different file is a recovery that can read anything on the disk, and the user chose those files.

PyHydroGeophysX.agents.runtime.recovery.MAX_RECOVERIES = 1

How many times one tool may be recovered within a run. One retry catches the unsuitable-setting case; a tool that fails twice with different settings is failing for a reason settings will not fix, and going round again only spends the user’s time and tokens.

PyHydroGeophysX.agents.runtime.recovery.MAX_SCRIPT_FIXES = 1

Fixes the model may write for an approved script that failed, each put to the user again before it runs.

PyHydroGeophysX.agents.runtime.recovery.SCRIPT_TOKENS = None

none. A capped reply cut every proposal with code in it short, mid-function.

Type:

The reply cap for a script

PyHydroGeophysX.agents.runtime.recovery.apply_changes(ctx: RunContext, changes: Dict[str, Any]) → List[str]

Put the diagnosis’s settings into the configuration, and say which.

Returns:

"key: old -> new" per change, for the transcript. The run’s configuration is what the report prints, so a number that changed mid-run has to be visible as having changed.

Return type:

list of str

Examples

>>> ctx = RunContext('goal', {'inversion_params': {'lambda': 15}})
>>> apply_changes(ctx, {'inversion_params': {'lambda': 5}})
["inversion_params: {'lambda': 15} -> {'lambda': 5}"]

PyHydroGeophysX.agents.runtime.recovery.ask_long(ask: Callable[[...], str], prompt: str, max_tokens: int | None = None) → str

Ask with no cap on the reply, when ask lets the caller say so.

PyHydroGeophysX.agents.runtime.recovery.diagnose(ctx: RunContext, tool: Any, error: str, ask: Callable[[str], str] | None) → Dict[str, Any]

Ask what went wrong and what to change, and keep only what is allowed.

Parameters:

• ctx (RunContext) – The run, for the transcript and the current settings.

• tool (Tool or None) – The tool that failed.

• error (str) – Its error message.

• ask (callable or None) – Takes a prompt, returns the model’s reply. Without one there is no diagnosis: a deterministic run does not guess at causes.

Returns:

{"cause", "action", "changes", "why"}. action is "retry" only when a diagnosis asked for it and named at least one allowed setting that would actually differ; otherwise "give_up", so a model that answers “retry” with nothing to change cannot loop the run.

Return type:

dict

Raises:

None – An unreachable or unparseable model gives up, which is what the loop did before any of this existed.

Examples

>>> ctx = RunContext('goal', {'inversion_params': {'lambda': 15}})
>>> reply = '{"cause": "lambda too high", "action": "retry",' \
...         ' "changes": {"inversion_params": {"lambda": 5}},' \
...         ' "why": "less smoothing"}'
>>> plan = diagnose(ctx, None, 'diverged', lambda prompt: reply)
>>> plan['action'], plan['changes']
('retry', {'inversion_params': {'lambda': 5}})
>>> diagnose(ctx, None, 'diverged', None)['action']
'give_up'

PyHydroGeophysX.agents.runtime.recovery.input_paths(ctx: RunContext) → Dict[str, str]

The run’s input files and folders that exist, by configuration key.

A list (time-lapse surveys) is spelled out one entry per file, as key[0], key[1]… A temporary script reads these and nothing else.

Examples

>>> input_paths(RunContext('goal', {'data_file': 'nowhere.dat', 'lambda': 5}))
{}

PyHydroGeophysX.agents.runtime.recovery.recover(ctx: RunContext, tool: Any, error: str, ask: Callable[[str], str] | None, run: Callable[[str], str], attempts: Dict[str, int] | None = None) → Tuple[str, str]

Diagnose a failed step and, if a setting can be blamed, run it again.

Parameters:

• ctx (RunContext) – The run. Its configuration is what a recovery changes.

• tool (Tool) – The tool that failed.

• error (str) – Its error.

• ask (callable or None) – The model, for the diagnosis.

• run (callable) – Takes the tool’s name and runs it again, returning the new status. The controller passes its own invoke so the retry is an ordinary step, recorded like any other.

• attempts (dict, optional) – Tool name to recoveries already spent, carried across a run.

Returns:

(status, note) - the status after any retry, and a sentence for the warnings. status is the original "failed" when nothing was tried.

Return type:

tuple

Raises:

None –

Examples

>>> from .tools import Tool
>>> ctx = RunContext('goal', {'inversion_params': {'lambda': 15}})
>>> tool = Tool('invert', 'Invert.', lambda c: ('', {}),
...             label='Run ERT inversion')
>>> reply = '{"cause": "too smooth", "action": "retry",' \
...         ' "changes": {"inversion_params": {"lambda": 5}}, "why": "ok"}'
>>> recover(ctx, tool, 'diverged', lambda p: reply, lambda name: 'ok')
('ok', "Run ERT inversion failed (diverged). Diagnosis: too smooth. Retried with inversion_params: {'lambda': 15} -> {'lambda': 5}, which worked.")

PyHydroGeophysX.agents.runtime.recovery.script_fix(ctx: RunContext, tool: Any, error: str, plan: Dict[str, Any], ask: Callable[[...], str] | None, run: Callable[[str], str]) → Tuple[str, str]

Have the model write a temporary script, and run it if the user says yes.

The model writes it; scratch.check_script() checks it (once more after a fix if the first draft breaks a rule); the user reads it and decides; and only then does it run, in a process and folder of its own (scratch.run_script()). When it stands in for an input (plan["replaces"]), the configuration points at its output and the step runs again. Declining is the default and a perfectly good outcome.

Returns:

(status, note) as recover() returns them.

Return type:

tuple

PyHydroGeophysX.agents.runtime.recovery.script_task(ctx: RunContext, task: str, ask: Callable[[...], str] | None, why: str = '') → str

A temporary script the controller asks for on its own initiative.

For something no step does - a quantity the tools do not compute, a check of a product against the data, a figure the goal needs. The model writes it, the user reads it with the reason and decides, and only then does it run. Returns a sentence for the transcript.

PyHydroGeophysX.agents.runtime.recovery.settings_for(ctx: RunContext) → Dict[str, Any]

The adjustable part of the configuration, as the diagnosis sees it.

PyHydroGeophysX.agents.runtime.scratch module

Temporary scripts an assistant writes to get past a problem, run only on a yes.

Some failures are neither a setting nor a missing feature: a file in a layout no reader handles, a model folder written in one module’s format that another module cannot open, a quantity none of the tools computes. A person at the keyboard would write ten lines of Python and carry on. This lets the assistant do the same - and only that far:

• The user decides. A script is shown with what it is for, and runs only after an explicit yes. A run with nobody to ask declines.

• The source is checked before anyone is asked (check_script()): one def main(inputs, out), imports from a short list of scientific libraries and this package, no exec/eval, no dunder attributes, no os, sys, subprocess or network modules.

• It runs in a process of its own (run_script()), in a folder of its own, with a time limit and without the API keys in its environment. Inside that process an audit hook refuses subprocesses, sockets and native-library loading, and any write outside the script’s folder (which holds its temporary directory too). The user’s files are read, never written.

That is a guard rail, not a security boundary: compiled code inside numpy or pyGIMLi does not pass through Python’s audit hooks. It keeps a well-meant script from doing something nobody approved; the approval is what stands between a script and the user’s data.

The script, its inputs, what it printed and what it wrote are kept in its folder, so the report can say what ran and the user can run it again.

PyHydroGeophysX.agents.runtime.scratch.ALLOWED_MODULES = frozenset({'PyHydroGeophysX', 'bisect', 'cmath', 'collections', 'copy', 'csv', 'dataclasses', 'datetime', 'decimal', 'fractions', 'functools', 'heapq', 'itertools', 'json', 'math', 'matplotlib', 'numpy', 'operator', 'pandas', 'pathlib', 'pygimli', 're', 'scipy', 'statistics', 'string', 'textwrap', 'typing'})

Top-level modules a script may import. Scientific libraries and this package; nothing that starts processes, opens sockets or edits the system.

PyHydroGeophysX.agents.runtime.scratch.DEFAULT_TIMEOUT_S = 180

Seconds a script may run before it is stopped.

PyHydroGeophysX.agents.runtime.scratch.FORBIDDEN_NAMES = frozenset({'__import__', 'breakpoint', 'compile', 'delattr', 'eval', 'exec', 'getattr', 'globals', 'help', 'input', 'locals', 'setattr', 'vars'})

each one is a way round the checks.

Type:

Built-in names a script may not call

PyHydroGeophysX.agents.runtime.scratch.MAX_SOURCE_CHARS = 6000

Longer than this and nobody reads it before saying yes.

PyHydroGeophysX.agents.runtime.scratch.SCRIPT_CONTRACT = 'Write Python with one function `def main(inputs, out):`. `inputs` is a dict of name -> file or folder path (strings) to read; `out` is a pathlib.Path folder, the only place you may write. Return a small JSON-serialisable dict saying what you found or did and naming the files you wrote (relative to `out`). Imports only from: numpy, scipy, pandas, matplotlib, pygimli, PyHydroGeophysX, math, statistics, json, csv, re, datetime, collections, itertools, functools, pathlib, typing, dataclasses. No os, sys, subprocess, shutil or network; no exec, eval, getattr or dunder attributes; never modify the input files. Keep it short - under 6000 characters - so a person can read it before approving it.'

What the assistant is told a script looks like. Shared by the chat and the automatic run, so both write scripts the same checks accept.

PyHydroGeophysX.agents.runtime.scratch.check_script(source: str) → str

Why source may not run, or “” when it may be offered to the user.

Parameters:

source (str) – The script as the model wrote it.

Returns:

A sentence the model can act on to fix the script, or “”.

Return type:

str

Raises:

None –

Examples

>>> check_script("import numpy as np\ndef main(inputs, out):\n    return {}")
''
>>> check_script("import os\ndef main(inputs, out):\n    os.remove('x')")
'It imports os, which a temporary script may not use.'
>>> check_script("def main(inputs, out):\n    return ().__class__")
'It uses __class__; dunder attributes are not allowed.'
>>> check_script("print('hi')")
'It must define main(inputs, out).'

PyHydroGeophysX.agents.runtime.scratch.describe_for_user(purpose: str, source: str, inputs: Mapping[str, Any]) → str

The question put to the user: what the script is for, what it reads, the code.

The prose comes first and the code last, so the studio can show the code in a box of its own (qt_apps.modules.base._split_code).

PyHydroGeophysX.agents.runtime.scratch.imported_modules(source: str) → List[str]

The modules source imports, to load before the write guard goes up.

PyHydroGeophysX.agents.runtime.scratch.run_script(source: str, inputs: Mapping[str, Any], out_dir: Any, purpose: str = '', timeout: float = 180) → Dict[str, Any]

Run an approved script in a process of its own and report what it did.

Call only after the user has said yes: this does not ask.

Parameters:

• source (str) – The script; checked again here.

• inputs (mapping) – Name -> path given to main as inputs. Every path must exist.

• out_dir (str or Path) – The script’s folder, created if needed; main gets it as out.

• purpose (str, optional) – What the script is for, kept with it.

• timeout (float, optional) – Seconds before the script is stopped.

Returns:

status (“ok” or “failed”), result (what main returned), files (written in out, relative), printed (the end of its output), error, folder and seconds.

Return type:

dict

Raises:

None –

PyHydroGeophysX.agents.runtime.scratch.script_folder(root: Any, label: str = 'script') → Path

A new folder for one script under root, named by time.

PyHydroGeophysX.agents.runtime.scratch.split_answer(reply: Any) → Tuple[str, str]

(purpose, code) from a model’s JSON proposal, or ("", "").

PyHydroGeophysX.agents.runtime.steering module

What the user can tell a run while it works: hold it, let it go on, and notes.

A run in the studio happens in its own process, and the person watching it may want to hold it before the next step, or tell it something - “use lambda 20”, “leave the seismic line out”, “write the report now”. This is the run’s side of that. The process sets a Steering for the run (current), fills it from another thread as messages arrive, and the controller reads it between steps (run_controller()):

• Pause takes effect before the next step. The step that is running runs to the end: an inversion stopped half-way leaves nothing anyone can use.

• A note reaches the controller with its next decision. It is added to the transcript the model reads, and that decision may change the run’s adjustable settings (ADJUSTABLE) to follow it - never a file, never code. A note the run cannot follow is answered with why, and a run with no model to read notes says so rather than pretending to.

Nothing here knows about Qt; Steering.notify is how the run tells whoever is listening that it paused, went on, or read a note, and Steering.source is where it reads what it is told - for the desktop, a ControlFile in the run’s folder.

class PyHydroGeophysX.agents.runtime.steering.ControlFile(path: str)

Bases: object

Messages appended to a file, each read once, in order.

How the desktop talks to a run in another process. A thread reading the run’s stdin for them froze runs on Windows: while one thread is blocked reading a pipe, starting a process - which a step may do - waits for that read to return. The desktop appends one JSON object per line instead, and the run reads the new lines between steps; an incomplete last line waits for the next read.

Examples

>>> import os, tempfile
>>> path = os.path.join(tempfile.mkdtemp(), 'steering.jsonl')
>>> read = ControlFile(path)
>>> read()
[]
>>> with open(path, 'a', encoding='utf-8') as f:
...     _ = f.write('{"steer": "use lambda 20"}\n{"pause": tr')
>>> read()
[{'steer': 'use lambda 20'}]
>>> with open(path, 'a', encoding='utf-8') as f:
...     _ = f.write('ue}\n')
>>> read(), read()
([{'pause': True}], [])

class PyHydroGeophysX.agents.runtime.steering.Steering(notify: Callable[[Dict[str, Any]], None] | None = None, source: Callable[[], List[Dict[str, Any]]] | None = None)

Bases: object

Pause, resume and notes for one run, safe to fill from another thread.

Parameters:

• notify (callable, optional) – Receives an event dict - phase "paused", "resumed" or "steer" - whenever the run acts on what it was told. Errors it raises are ignored: a display must not stop a run.

• source (callable, optional) – Returns the messages that have arrived since it was last called - {"steer": text}, {"pause": true}, {"resume": true}, {"stop": true}. Read at every checkpoint, and while paused.

Examples

>>> seen = []
>>> steer = Steering(notify=seen.append)
>>> steer.say('use lambda 20')
>>> steer.take_notes(), steer.take_notes()
(['use lambda 20'], [])
>>> steer.checkpoint()          # not paused: carries straight on
True
>>> steer.stop()
>>> steer.checkpoint()
False

POLL_S = 0.25

How often a paused run looks for a resume, in seconds.

checkpoint(where: str = '') → bool

Wait here while paused. False when the run should stop instead.

pause() → None

Hold the run before its next step.

property paused: bool

poll() → None

Take in whatever source has received since last time.

receive(message: Dict[str, Any]) → None

Act on one message from source.

resume() → None

say(text: str) → None

Queue a note for the controller’s next decision.

stop() → None

End the run at its next checkpoint - the listener has gone.

take_notes() → List[str]

The notes not yet read, oldest first; each is returned once.

tell(event: Dict[str, Any]) → None

PyHydroGeophysX.agents.runtime.steering.current: ContextVar = <ContextVar name='steering' default=None>

The steering of the run in progress in this context, or None - a script, a test, or a run nobody is watching, which then runs as it always did.

PyHydroGeophysX.agents.runtime.studio module

The desktop studio’s pages, as tools an automatic run can use.

An automatic run works headless in a process of its own, and its tools are the runtime catalog’s. The studio’s pages hold more: ERT quality control, mesh zones, other inversion engines, temperature correction, seismic picking, boreholes, the Project Map. This is how a run reaches them - the same page actions the user clicks and the chat drives (agent_describe / agent_apply), carried out by the desktop on its own thread (qt_apps.agent.controller.StudioController.serve_studio).

The desktop sets current to a function that takes one request and returns the answer; elsewhere (a script, the web app, a test) it is None and the run is offered no studio pages. A request is {"op": ...}:

• catalog - the pages a run may use, with what each is for;

• describe - one page’s actions, their arguments and its state;

• apply - run one action; a long one (an inversion) is answered when it has finished;

• handover - the page’s finished result, written out for the run (the ERT page’s model, see take_studio_inversion in the catalog).

PyHydroGeophysX.agents.runtime.studio.available() → bool

Whether this run can use the studio’s pages.

PyHydroGeophysX.agents.runtime.studio.call(op: str, **request: Any) → Dict[str, Any]

Send one request to the studio and return its answer.

Never raises: no studio, or a bridge that fails, is a failed answer the caller reports like any other.

Examples

>>> call('describe', module='ert')['status']
'failed'

PyHydroGeophysX.agents.runtime.studio.catalog() → List[Dict[str, str]]

The pages a run may use, asked once per process.

PyHydroGeophysX.agents.runtime.studio.current: ContextVar = <ContextVar name='studio_bridge' default=None>

request -> answer. None when there is no studio.

Type:

The desktop’s bridge

PyHydroGeophysX.agents.runtime.studio.describe_compact(desc: Any, action: str = '') → str

A page’s description, short enough for the controller to read.

Every action with its arguments and the first sentence of what it does, then the page’s state - a page with twenty actions and paragraphs of detail on each would otherwise crowd out everything else the model reads. With action, that one action’s whole description instead.

Examples

>>> page = {'actions': [{'name': 'run', 'args': {'lambda': 'float'},
...                      'desc': 'Invert. Long detail.'}], 'state': {'has_model': False}}
>>> print(describe_compact(page))
actions: run(lambda: float) - Invert.
state: {'has_model': False}
>>> print(describe_compact(page, 'run'))
run(lambda: float): Invert. Long detail.

PyHydroGeophysX.agents.runtime.studio.result_summary(answer: Dict[str, Any], limit: int = 700) → str

What a page action came back with, in one line for the transcript.

PyHydroGeophysX.agents.runtime.toolbox module

What the controller may do besides the next step on the standard route.

The registry’s dependency order is the standard route: load, invert, evaluate, convert, report. It is a good default and it stays the default - a run without a model follows it exactly, and a run with one follows it unless the evidence says otherwise. What it cannot do on its own is react to that evidence beyond the one retry of a failed step: run the inversion again because the fit is poor, add the climate series that would explain a change, look at a product’s numbers before deciding, or write ten lines of Python for something no tool does.

This module is that latitude, and its limits:

• Read-only looks are free. inspect summarises a product’s numbers into the transcript; it changes nothing, so it needs nobody’s permission.

• Leaving a step out is the model’s call. A step for data the request never named, or a method it did not ask for, is dropped on the model’s judgement, with its reason recorded - but never one the report or a requested product still needs (controller._skip_steps).

• Every other change of course is the user’s decision. A tool the route has not scheduled, a step run again with other settings, a changed setting, a temporary script, a question: each is put to the user with the model’s reason as its recommendation (propose()), and staying on the route is the default - the answer when nobody is there.

• Bounded. At most MAX_PROPOSALS proposals in a run, and a declined proposal is not offered again, so a run cannot turn into a negotiation.

PyHydroGeophysX.agents.runtime.toolbox.MAX_INSPECTIONS = 10

Looks a run may take - at a product’s numbers, or at a studio page’s actions and state - before it must act on what it has seen.

PyHydroGeophysX.agents.runtime.toolbox.MAX_PROPOSALS = 4

Off-route proposals a run may put to the user. Each costs the user an answer; a handful is enough to change course where it matters.

PyHydroGeophysX.agents.runtime.toolbox.ask_choice(ctx: RunContext, question: str, options: Sequence[str], recommend: str = '', why: str = '') → str

Ask the user a question of the model’s, and record the answer as guidance.

Examples

>>> ctx = RunContext('x')
>>> ask_choice(ctx, 'Archie or Waxman-Smits?', ['Archie', 'Waxman-Smits'], 'Archie')
'Archie'
>>> print(ctx.guidance[-1])
You asked: Archie or Waxman-Smits? The user chose: Archie (the default; nobody answered).

PyHydroGeophysX.agents.runtime.toolbox.declined(ctx: RunContext) → Set[str]

Keys of the proposals the user turned down, never to be offered again.

PyHydroGeophysX.agents.runtime.toolbox.describe_value(value: Any, depth: int = 0, budget: int = 1600) → str

A product’s numbers in a few lines: shapes, ranges, the scalars it holds.

Examples

>>> describe_value(np.array([[1.0, 2.0], [3.0, np.nan]]))
'array (2, 2): 3 finite, min 1, median 2, max 3'
>>> describe_value({'chi2': 1.5, 'status': 'success'})
'{chi2: 1.5; status: success}'

PyHydroGeophysX.agents.runtime.toolbox.downstream(tools: Dict[str, Any], produced: Sequence[str]) → List[str]

Every tool that reads, directly or through others, what produced names.

Examples

>>> from .tools import Tool
>>> reg = {'inv': Tool('inv', '', lambda c: ('', {}), produces=('m',)),
...        'ev': Tool('ev', '', lambda c: ('', {}), requires=('m',), produces=('e',)),
...        'rep': Tool('rep', '', lambda c: ('', {}), requires=('e',))}
>>> downstream(reg, ['m'])
['ev', 'rep']

PyHydroGeophysX.agents.runtime.toolbox.extra_tools(ctx: RunContext, tools: Dict[str, Any]) → List[Any]

Tools whose inputs exist but which the route has not scheduled.

Their when gate is closed - the request did not name their product, or the configuration does not call for them - and they have not been tried. The gate is right for a run that cannot ask; with a user to ask, the model may propose one.

Examples

>>> from .tools import Tool
>>> only = {'climate': Tool('climate', 'Weather.', lambda c: ('', {}),
...                         when=lambda c: False)}
>>> [t.name for t in extra_tools(RunContext('x'), only)]
['climate']

PyHydroGeophysX.agents.runtime.toolbox.inspect_artifact(ctx: RunContext, key: str) → str

Record a read-only look at product key as a step, and return what it saw.

PyHydroGeophysX.agents.runtime.toolbox.looks_left(ctx: RunContext) → int

How many more looks - at a product, or at a studio page - this run may take.

PyHydroGeophysX.agents.runtime.toolbox.proposals_left(ctx: RunContext) → int

How many more off-route proposals this run may make.

PyHydroGeophysX.agents.runtime.toolbox.propose(ctx: RunContext, key: str, title: str, why: str, route_next: str, name: str = 'The assistant', detail: str = '') → bool

Put an off-route action to the user, recommending it; True if accepted.

key identifies the proposal (extra:<tool>, redo:<tool>, settings, script…) so a declined one is never offered again. Staying on the route is the default and the answer when nobody is there.

Examples

>>> ctx = RunContext('x')
>>> propose(ctx, 'extra:climate', 'Fetch the climate series', 'rain explains it',
...         'Write the report')
False
>>> ctx.proposals[0]['accepted'], 'extra:climate' in declined(ctx)
(False, True)
>>> ctx.settings['ask_user'] = lambda event: 'go'
>>> propose(ctx, 'redo:invert', 'Invert again with lambda 20', 'chi2 is 39', 'Evaluate')
True

PyHydroGeophysX.agents.runtime.toolbox.redo_candidates(ctx: RunContext, tools: Dict[str, Any], finish: str = 'report_files') → List[Any]

Finished steps that could be run again with other settings.

Not the step that writes the final product: it is run again by itself whenever a step before it is (downstream()).

PyHydroGeophysX.agents.runtime.toolbox.skippable(ctx: RunContext, tools: Dict[str, Any], route: Sequence[str], finish: str = 'report_files') → List[str]

Steps on the route ahead that may be left out, by name.

Not the step that writes the final product, and not one already tried. The steps after a skipped one that need its products drop off the route by themselves. Whether a step is still needed for a requested product is decided when it is skipped (controller._skip_steps).

Examples

>>> from .tools import Tool
>>> reg = {'tem': Tool('tem', '', lambda c: ('', {}), produces=('t',)),
...        'rep': Tool('rep', '', lambda c: ('', {}), produces=('report_files',))}
>>> skippable(RunContext('x'), reg, ['tem', 'rep'])
['tem']

PyHydroGeophysX.agents.runtime.toolbox.studio_pages_allowed(ctx: RunContext) → Set[str]

The studio pages the user has let this run use.

PyHydroGeophysX.agents.runtime.toolbox.studio_text(ctx: RunContext) → List[str]

The studio’s pages, as the controller’s prompt lists them; [] without a studio.

PyHydroGeophysX.agents.runtime.toolbox.toolbox_text(ctx: RunContext, tools: Dict[str, Any], scripts: bool = True, route: Sequence[str] = (), finish: str = 'report_files') → str

The off-route actions as the controller’s prompt lists them.

route is the standard route ahead, by step name, for the steps the model may propose to leave out.

PyHydroGeophysX.agents.runtime.tools module

Every capability the workflow has, described well enough to be chosen.

A tool is the unit the controller selects. It is deliberately not the same thing as an agent class: an agent is an implementation, a tool is an offer - what it needs, what it produces, and one sentence of what it is for. The controller never sees ERTInversionAgent; it sees “invert_time_lapse: recover a resistivity model for every survey at once, needs ert_data”.

Three fields carry the weight:

requires

Artifacts that must exist. A tool whose requirements are unmet is never offered, which is what makes an impossible sequence unrepresentable rather than merely discouraged. The old pipeline enforced ordering by writing the calls in order; here the data dependencies say it.

produces

What will exist afterwards, so the controller can reason about reaching a goal it has not been given a recipe for.

description

Written for a reader who does not know this codebase, because that is what the model is. This is the same reason a tool description matters more than a function docstring: nothing else tells the caller when to reach for it.

Handlers take the RunContext and return (summary, outputs). Raising is allowed and is not fatal - the controller records the failure as an observation and carries on deciding.

PyHydroGeophysX.agents.runtime.tools.TOOLS: Dict[str, Tool] = {'convert_mt_water_content': Tool(name='convert_mt_water_content', description="Convert each MT site's layered resistivity model to volumetric water content, layer by layer, propagating petrophysical uncertainty by Monte Carlo. Required when the request asks about water content and the MT sites are the run's only resistivity models.", handler=<function _convert_mt_water_content>, requires=('mt_results', 'mt_evaluation'), produces=('water_content',), agent='PetrophysicsAgent', label='Convert MT models to water content', module='mt', when=<function _mt_water_content_wanted>, repeatable=False, proposable=True, applies=None), 'convert_tdem_water_content': Tool(name='convert_tdem_water_content', description="Convert the TDEM sounding's layered resistivity model to volumetric water content, layer by layer, propagating petrophysical uncertainty by Monte Carlo. Required when the request asks about water content and the sounding is the run's only resistivity model.", handler=<function _convert_tdem_water_content>, requires=('tdem_results', 'tdem_evaluation'), produces=('water_content',), agent='PetrophysicsAgent', label='Convert TDEM model to water content', module='em', when=<function _tdem_water_content_wanted>, repeatable=False, proposable=True, applies=None), 'convert_water_content': Tool(name='convert_water_content', description='Convert the recovered resistivity to volumetric water content for every survey, propagating petrophysical uncertainty by Monte Carlo. Required whenever the request asks about water content, moisture or saturation.', handler=<function _convert_water_content>, requires=('inversion_results', 'evaluation_results'), produces=('water_content',), agent='PetrophysicsAgent', label='Convert to water content', module='geo_hydrology', when=<function <lambda>>, repeatable=False, proposable=True, applies=None), 'derive_structure': Tool(name='derive_structure', description='Turn a seismic velocity model into layer interfaces that can constrain an ERT inversion or a petrophysical conversion.', handler=<function _derive_structure>, requires=('seismic_results', 'seismic_structure', 'ert_data', 'inversion_results'), produces=('structure_results',), agent='StructureConstraintAgent', label='Extract structural constraints', module='seismic3d', when=None, repeatable=False, proposable=True, applies=None), 'evaluate_gravmag_inversion': Tool(name='evaluate_gravmag_inversion', description="Judge how well the 3D density-contrast or susceptibility model explains the residual anomaly, whether it reaches the solver's bounds, and whether the beta search landed on its target. Run after the gravity / magnetic inversion.", handler=<function _evaluate_gravmag>, requires=('gravmag_results',), produces=('gravmag_evaluation',), agent='GravMagAgent', label='Evaluate gravity / magnetic inversion', module='gravmag', when=<function _gravmag_inverted>, repeatable=False, proposable=True, applies=None), 'evaluate_inversion': Tool(name='evaluate_inversion', description='Judge how well the recovered model fits the data and whether it is physically plausible, retrying with adjusted regularization if the configuration allows. Run after any inversion.', handler=<function _evaluate>, requires=('inversion_results',), produces=('evaluation_results',), agent='InversionEvaluationAgent', label='Evaluate inversion quality', module='ert', when=None, repeatable=False, proposable=True, applies=None), 'evaluate_mt_inversion': Tool(name='evaluate_mt_inversion', description="Judge how well each MT site's model fits its data against Occam's target RMS, how large the static shifts are, and whether the resistivities are physical. Run after the MT inversion, before anything is built on it.", handler=<function _evaluate_mt>, requires=('mt_results',), produces=('mt_evaluation',), agent='MT inversion (Occam 1D, SimPEG 2D)', label='Evaluate MT inversion', module='mt', when=None, repeatable=False, proposable=True, applies=None), 'evaluate_seismic_inversion': Tool(name='evaluate_seismic_inversion', description="Judge how well the velocity model fits the travel times, how much of it the rays cover and whether velocities sit at the inversion's bounds; re-invert with lambda changed while the fit is off. Run after the seismic inversion, before anything is built on the model.", handler=<function _evaluate_seismic>, requires=('seismic_results',), produces=('seismic_evaluation',), agent='SeismicAgent', label='Evaluate seismic inversion', module='seismic', when=None, repeatable=False, proposable=True, applies=None), 'evaluate_tdem_inversion': Tool(name='evaluate_tdem_inversion', description="Judge how well the TDEM model fits each sounding's decay, how many soundings resolved a model above their depth of investigation, and whether its resistivities are physical. Run after the TDEM inversion, before anything is built on the model.", handler=<function _evaluate_tdem>, requires=('tdem_results',), produces=('tdem_evaluation',), agent='TDEMAgent', label='Evaluate TDEM inversion', module='em', when=None, repeatable=False, proposable=True, applies=None), 'extract_seismic_interfaces': Tool(name='extract_seismic_interfaces', description='Trace layer interfaces through the seismic velocity model at the threshold velocities (velocity_threshold, 1200 m/s by default), and draw the model with them.', handler=<function _extract_seismic_interfaces>, requires=('seismic_results', 'seismic_evaluation'), produces=('seismic_structure',), agent='SeismicAgent', label='Extract layer interfaces', module='seismic', when=None, repeatable=False, proposable=True, applies=None), 'fetch_climate': Tool(name='fetch_climate', description="Retrieve precipitation, temperature and potential evapotranspiration for the survey dates at the site's coordinates. Only useful when the request asks to relate the geophysics to weather.", handler=<function _fetch_climate>, requires=(), produces=('climate_data',), agent='ClimateDataAgent', label='Fetch climate data', module='geo_hydrology', when=<function _climate_wanted>, repeatable=False, proposable=True, applies=None), 'fuse_methods': Tool(name='fuse_methods', description='Combine two or more geophysical methods into a single interpretation. Only useful when more than one method has produced a result.', handler=<function _fuse>, requires=(), produces=('fusion_results',), agent='DataFusionAgent', label='Fuse methods', module='joint_inversion', when=<function <lambda>>, repeatable=False, proposable=True, applies=None), 'invert_ert': Tool(name='invert_ert', description='Recover a resistivity model from a single ERT survey. Use when exactly one survey is loaded.', handler=<function _invert_single>, requires=('ert_data',), produces=('inversion_results',), agent='ERTInversionAgent', label='Run ERT inversion', module='ert', when=<function _is_single>, repeatable=False, proposable=True, applies=<function <lambda>>), 'invert_gravmag': Tool(name='invert_gravmag', description='Process gravity or magnetic station data: separate the regional trend from the residual anomaly, map both, and invert the residual for a 3D density-contrast or susceptibility model.', handler=<function _invert_gravmag>, requires=('gravmag_data',), produces=('gravmag_results',), agent='GravMagAgent', label='Run gravity / magnetic inversion', module='gravmag', when=<function _configured.<locals>.gate>, repeatable=False, proposable=True, applies=None), 'invert_mt': Tool(name='invert_mt', description='Invert magnetotelluric sites (EDI, EMTF XML, Z- or J-files) for layered resistivity by Occam 1D, and a line of three or more located sites for a 2D TE/TM section.', handler=<function _invert_mt>, requires=('mt_data',), produces=('mt_results',), agent='MT inversion (Occam 1D, SimPEG 2D)', label='Run MT inversion', module='mt', when=<function _configured.<locals>.gate>, repeatable=False, proposable=True, applies=None), 'invert_seismic': Tool(name='invert_seismic', description='Invert first-arrival travel times for a P-wave velocity model by refraction tomography.', handler=<function _invert_seismic>, requires=(), produces=('seismic_results',), agent='SeismicAgent', label='Run seismic refraction inversion', module='seismic', when=<function <lambda>>, repeatable=False, proposable=False, applies=None), 'invert_tdem': Tool(name='invert_tdem', description='Invert a time-domain electromagnetic sounding for a layered resistivity model.', handler=<function _invert_tdem>, requires=('tdem_data',), produces=('tdem_results',), agent='TDEMAgent', label='Run TDEM inversion', module='em', when=<function _configured.<locals>.gate>, repeatable=False, proposable=True, applies=None), 'invert_time_lapse': Tool(name='invert_time_lapse', description='Recover a resistivity model for every survey at once, with a temporal constraint linking consecutive surveys. Use when two or more surveys are loaded and the question is how the subsurface changed.', handler=<function _invert_time_lapse>, requires=('ert_data',), produces=('inversion_results',), agent='ERTInversionAgent', label='Run time-lapse inversion', module='ert', when=<function _is_time_lapse>, repeatable=False, proposable=True, applies=<function <lambda>>), 'load_ert_surveys': Tool(name='load_ert_surveys', description='Read the ERT survey files named in the configuration, applying electrode geometry and topography. Run this before any ERT inversion.', handler=<function _load_ert>, requires=(), produces=('ert_data',), agent='ERTLoaderAgent', label='Load ERT data', module='ert', when=<function _configured.<locals>.gate>, repeatable=False, proposable=False, applies=None), 'load_gpr_line': Tool(name='load_gpr_line', description='Read the ground-penetrating radar line the configuration names (GSSI DZT, pulseEKKO DT1/HD, MALA rd3/rd7/rad, SEG-Y or .npz) and draw its radargram as recorded. Run before the GPR processing.', handler=<function _load_gpr>, requires=(), produces=('gpr_data',), agent='data_processing.gpr', label='Load GPR line', module='gpr', when=<function _configured.<locals>.gate>, repeatable=False, proposable=False, applies=None), 'load_gravmag_data': Tool(name='load_gravmag_data', description="Read the gravity or magnetic station table the configuration names, say which field it holds, and map the stations' values. Run before the gravity / magnetic processing.", handler=<function _load_gravmag>, requires=(), produces=('gravmag_data',), agent='GravMagAgent', label='Load gravity / magnetic data', module='gravmag', when=<function _configured.<locals>.gate>, repeatable=False, proposable=False, applies=None), 'load_model_output': Tool(name='load_model_output', description='Read water content or saturation from a hydrological model such as MODFLOW or ParFlow, for comparison with the geophysics.', handler=<function _load_model_output>, requires=(), produces=('model_output',), agent='ModelOutputAgent', label='Load hydrological model output', module='hydro_geophysics', when=<function _configured.<locals>.gate>, repeatable=False, proposable=False, applies=None), 'load_mt_sites': Tool(name='load_mt_sites', description="Read the MT transfer functions the configuration names and draw each site's apparent resistivity and phase. Run before the MT inversion.", handler=<function _load_mt>, requires=(), produces=('mt_data',), agent='MT inversion (Occam 1D, SimPEG 2D)', label='Load MT sites', module='mt', when=<function _configured.<locals>.gate>, repeatable=False, proposable=False, applies=None), 'load_seismic_traveltimes': Tool(name='load_seismic_traveltimes', description='Read the first-arrival travel-time file the configuration names and draw its travel-time curves. Run before the seismic inversion.', handler=<function _load_seismic_traveltimes>, requires=(), produces=('seismic_traveltimes',), agent='SeismicAgent', label='Load seismic travel times', module='seismic', when=<function <lambda>>, repeatable=False, proposable=False, applies=None), 'load_studio_inversion': Tool(name='load_studio_inversion', description='Take over the ERT inversion the user already ran step by step in the studio, so the run carries on from that model instead of inverting again. Offered only when a step-by-step task was handed to the run; use it before any inversion.', handler=<function _load_studio_inversion>, requires=('ert_data',), produces=('inversion_results',), agent='', label='Take over the studio inversion', module='ert', when=<function _studio_inversion_pending>, repeatable=False, proposable=False, applies=None), 'load_tdem_data': Tool(name='load_tdem_data', description='Read the TDEM sounding or survey the configuration names and draw the decay curves of a few soundings. Run before the TDEM inversion.', handler=<function _load_tdem>, requires=(), produces=('tdem_data',), agent='TDEMAgent', label='Load TDEM data', module='em', when=<function _configured.<locals>.gate>, repeatable=False, proposable=False, applies=None), 'map_tdem_plan_view': Tool(name='map_tdem_plan_view', description="Krige the TDEM survey's resistivity onto plan-view maps at several depths, filled across the survey outline, with a map of the kriging uncertainty for each, over the survey's basemap. Required when the request asks for the spatial distribution of a TEM survey.", handler=<function _map_tdem_plan_view>, requires=('tdem_results', 'tdem_evaluation'), produces=('plan_maps',), agent='TDEMAgent', label='Interpolate plan-view maps', module='em', when=<function _plan_maps_wanted>, repeatable=False, proposable=True, applies=None), 'pick_first_breaks': Tool(name='pick_first_breaks', description='Pick the first-arrival travel times in a raw SEG-Y file, place them on the survey geometry, and leave out shots whose times disagree with their reciprocals.', handler=<function _pick_first_breaks>, requires=(), produces=('seismic_picks',), agent='SeismicAgent', label='Pick first-arrival travel times', module='seismic', when=<function <lambda>>, repeatable=False, proposable=False, applies=None), 'process_gpr_line': Tool(name='process_gpr_line', description='Process the GPR line: dewow, time zero on the first arrival, background removal and gain (band-pass and migration when asked), convert time to depth with the velocity the request gives, and write the processed line (.npz), its radargram and the processing recipe (.json).', handler=<function _process_gpr>, requires=('gpr_data',), produces=('gpr_results',), agent='data_processing.gpr', label='Process GPR line', module='gpr', when=<function _configured.<locals>.gate>, repeatable=False, proposable=True, applies=None), 'take_studio_inversion': Tool(name='take_studio_inversion', description="Bring the inversion finished on the studio's ERT page into this run, for example one run there with quality control, zones or another engine, so this run evaluates, interprets and reports that model.", handler=<function _take_studio_inversion>, requires=('ert_data',), produces=('inversion_results',), agent='', label="Take over the ERT page's inversion", module='ert', when=<function <lambda>>, repeatable=False, proposable=True, applies=<function <lambda>>), 'write_report': Tool(name='write_report', description='Write the final report from everything produced so far, with its figures. Run this last, once the products the request asked for exist.', handler=<function _write_report>, requires=('inversion_results',), produces=('report_files',), agent='ReportAgent', label='Generate report', module='one_click', when=None, repeatable=False, proposable=True, applies=None), 'write_survey_report': Tool(name='write_survey_report', description='Write the final report of a run without ERT, from the TDEM, seismic, MT, gravity/magnetic and GPR results it produced: data, method, fit, the models, water content if converted, figures, recommendations. Run this last.', handler=<function _write_survey_report>, requires=(), produces=('report_files',), agent='ReportAgent', label='Generate report', module='one_click', when=<function _survey_report_due>, repeatable=False, proposable=False, applies=None)}

Every registered tool, by name.

class PyHydroGeophysX.agents.runtime.tools.Tool(name: str, description: str, handler: ~typing.Callable[[~PyHydroGeophysX.agents.runtime.context.RunContext], ~typing.Tuple[str, ~typing.Dict[str, ~typing.Any]]], requires: ~typing.Sequence[str] = <factory>, produces: ~typing.Sequence[str] = <factory>, agent: str = '', label: str = '', module: str = '', when: ~typing.Callable[[~PyHydroGeophysX.agents.runtime.context.RunContext], bool] | None = None, repeatable: bool = False, proposable: bool = True, applies: ~typing.Callable[[~PyHydroGeophysX.agents.runtime.context.RunContext], bool] | None = None)

Bases: object

One capability the controller can choose.

agent: str = ''

Which agent class implements it, for the derived execution plan.

applies: Callable[[RunContext], bool] | None = None

whether the tool applies to this run at all, whatever its gate says (a time-lapse inversion needs two surveys).

Type:

For an off-route proposal

available(ctx: RunContext) → bool

Whether this tool can run against the context as it stands.

description: str

handler: Callable[[RunContext], Tuple[str, Dict[str, Any]]]

label: str = ''

A short imperative label for the plan, e.g. “Run time-lapse inversion”.

missing(ctx: RunContext) → List[str]

Requirements that are not satisfied, for a blocked-step message.

module: str = ''

The desktop studio module that shows this kind of work, as keyed in PyHydroGeophysX.qt_apps.modules.MODULE_SPECS. The runtime never imports Qt; it reports the key and the desktop decides what to do with it, which is what lets a run walk the user through the studio while it works instead of leaving them on a progress bar.

name: str

offer() → str

The tool as one line of the controller’s menu.

produces: Sequence[str]

proposable: bool = True

Whether the controller may propose this tool off the standard route (toolbox) when its when gate is closed. Not for a loader, which reads a file the configuration has to name.

repeatable: bool = False

Tools that make no sense to run twice. The controller will not re-offer one that has already been attempted unless it is marked repeatable.

requires: Sequence[str]

unavailable_because(ctx: RunContext) → str

Why available() is False, in words the controller can act on.

when: Callable[[RunContext], bool] | None = None

Extra condition beyond requires - for tools gated by configuration rather than by data, such as the climate step.

PyHydroGeophysX.agents.runtime.tools.invoke(ctx: RunContext, name: str, reason: str = '', tools: Dict[str, Tool] | None = None) → str

Run one tool, recording the step whatever happens.

Only requires is checked here, not the when gate or the no-repeat rule: a recovery retries a tool that has just failed, which the no-repeat rule would refuse. Refusing a tool the menu did not offer is the controller’s job, before it gets this far.

Parameters:

• ctx (RunContext) – The run; the tool reads and writes it.

• name (str) – Registered tool name.

• reason (str) – Why this step was chosen, for the transcript and the plan.

Returns:

The step’s status: 'ok', 'failed', 'blocked' or 'skipped' (the last for an unknown tool name).

Return type:

str

Raises:

None – A tool that raises is recorded as failed. The loop has to keep its ability to decide what to do about a failure, which it loses if the exception propagates out of the run.

Examples

>>> ctx = RunContext('x')
>>> only = {'boom': Tool('boom', 'Raise.', lambda c: 1 / 0)}
>>> invoke(ctx, 'boom', tools=only)
'failed'
>>> ctx.steps[-1].error.startswith('ZeroDivisionError')
True
>>> invoke(ctx, 'no_such_tool', tools=only)
'skipped'

PyHydroGeophysX.agents.runtime.tools.menu(ctx: RunContext, tools: Dict[str, Tool] | None = None) → str

The runnable tools as the controller sees them, one per line.

PyHydroGeophysX.agents.runtime.tools.register(tool: Tool) → Tool

Add tool to the registry, replacing any tool of the same name.

Parameters:

tool (Tool) – The tool to register.

Returns:

The same tool, so this can wrap a definition.

Return type:

Tool

Raises:

ValueError – If the tool has no name or no handler - a nameless tool cannot be chosen and a handlerless one cannot be run, and both fail far from here.

Examples

>>> t = register(Tool('noop', 'Do nothing.', lambda ctx: ('done', {})))
>>> TOOLS['noop'] is t
True
>>> del TOOLS['noop']

PyHydroGeophysX.agents.runtime.tools.runnable_tools(ctx: RunContext, tools: Dict[str, Tool] | None = None) → List[Tool]

Every tool whose requirements the context currently satisfies.

Parameters:

ctx (RunContext) – The run as it stands.

Returns:

In registration order, which is roughly the order a run needs them - so a controller that simply takes the first is already a sane pipeline.

Return type:

list of Tool

Examples

>>> ctx = RunContext('x')
>>> only = {'needs_data': Tool('needs_data', 'Needs data.',
...                            lambda c: ('', {}), requires=('data',))}
>>> [t.name for t in runnable_tools(ctx, only)]
[]
>>> ctx.put('data', [1])
>>> [t.name for t in runnable_tools(ctx, only)]
['needs_data']

Module contents

A controller loop for the geophysics agents, in place of a fixed pipeline.

The workflow used to be a classifier followed by one of eight hard-coded sequences: _detect_workflow_type picked a branch of a 1770-line if/elif and that branch called its agents in a fixed order. Three things followed from that shape, and all three were reported as bugs rather than recognised as one design:

• The execution plan was written by hand in nine separate places, so it described the run someone expected rather than the run that happened. A plan listed ClimateDataAgent on a run whose configuration had climate disabled.

• BaseAgent.context is per-agent and never shared, so no agent could see what another had concluded. The report agent received the evaluation’s numbers but not its judgement, and could not say why a result needed review.

• Nothing could react. A stage that found a problem had no way to change what happened next, beyond one special case for re-running an inversion.

This package replaces the dispatcher with a controller loop: one transcript, one tool per capability, and a controller that reads the transcript, chooses the next action, observes its result, and chooses again.

• context is the transcript: the goal, every step taken, what each step found, and the artifacts they produced.

• tools is the registry: each capability described well enough for a model to choose it, with the preconditions that say when it can run at all.

• controller is the loop.

The loop degrades to a deterministic policy when there is no model to ask, so a run without an API key still executes - it simply stops adapting.

class PyHydroGeophysX.agents.runtime.RunContext(goal: str = '', config: Dict[str, Any] | None = None, output_dir: str = '', settings: Dict[str, Any] | None = None)

Bases: object

Everything one workflow run knows about itself.

Parameters:

• goal (str) – The user’s request, in their own words.

• config (dict, optional) – The parsed workflow configuration.

• output_dir (str, optional) – Where artifacts are written.

• settings (dict, optional) – Anything a tool needs that is not configuration - API key, model name, provider, progress callback.

Examples

>>> ctx = RunContext('invert this line', {'ert_file': 'a.dat'})
>>> ctx.put('ert_data', object(), 'Loaded 1 survey, 812 measurements.')
>>> ctx.has('ert_data')
True
>>> ctx.begin('invert_ert', 'the data are loaded').tool
'invert_ert'
>>> ctx.finish(summary='Converged to chi-squared 1.6.')
>>> ctx.steps[-1].status
'ok'

artifacts: Dict[str, Any]

ask(question: str, options: List[Dict[str, str]], default: str = '') → str

Put a choice to the user and wait, or take default if nobody is there.

For the cases a run genuinely cannot decide for itself: two files that disagree about their origin, a conversion with no calibration, a product that can be produced two defensible ways. Guessing there is worse than asking, and failing is worse than either - the run has usually done a lot of work by the time it finds out.

Parameters:

• question (str) – What is being asked, in one or two sentences, stating the evidence.

• options (list of dict) – {"id", "label", "detail"} per choice. id is what comes back; detail says what that choice will actually do.

• default (str) – The id to use when there is no one to ask - a headless run, a script, a test. Defaults to the first option.

Returns:

The chosen option’s id. Always one of the offered ids: an answer that matches nothing falls back to default, because a typo must not select behaviour nobody offered.

Return type:

str

Raises:

None –

Examples

>>> ctx = RunContext('goal')
>>> options = [{'id': 'a', 'label': 'Use the station file'},
...            {'id': 'b', 'label': 'Use the SEG-Y headers'}]
>>> ctx.ask('Which origin is right?', options)   # nobody to ask
'a'
>>> ctx.warnings[0].startswith('Nobody was available')
True
>>> ctx.settings['ask_user'] = lambda event: 'b'
>>> ctx.ask('Which origin is right?', options)
'b'

attempted(tool: str) → int

How many times tool has been tried, successfully or not.

A blocked step is not an attempt: the tool never ran, because something it needed was missing or it was not on offer. Counting it made a premature choice permanent - a conversion named before the evaluation existed was never offered again once the evaluation did exist.

begin(tool: str, reason: str = '', agent: str = '', description: str = '', purpose: str = '') → Step

Open a step. Only one is open at a time.

config: Dict[str, Any]

descriptions: Dict[str, str]

A one-line description of each artifact, for the transcript. The artifact itself may be a hundred megabytes; this is what the model gets to see about it.

elapsed() → float

ended

finished, exhausted, stopped or step_limit (see controller), “” until it has run. A run the user or the step limit cut off did not run out of work, and whether it is complete depends on which it was.

Type:

How the controller’s loop ended

finish(summary: str = '', status: str = 'ok', error: str = '') → None

Close the open step with what it concluded.

get(key: str, default: Any = None) → Any

guidance: List[str]

What the user told the run while it worked (steering), in order. Part of the transcript, so every later decision is made with it in view, not only the one that first read it.

has(*keys: str) → bool

True when every named artifact is present and not None.

note(message: str) → None

A warning the user should see, deduplicated.

plan() → List[Dict[str, Any]]

The execution plan, derived from what actually ran.

Returns the shape the report and the audit already consume, so a plan can no longer disagree with the run it describes.

projected(key: str) → bool

True when key exists only in a projected route, not yet in fact.

>>> ctx = RunContext('x')
>>> ctx.put('data', PROJECTED)
>>> ctx.has('data'), ctx.projected('data')
(True, True)

proposals: List[Dict[str, Any]]

what was proposed and whether it was accepted, in order.

Type:

Off-route actions put to the user (toolbox)

put(key: str, value: Any, description: str = '') → None

Record an artifact and how to describe it.

questions: List[Dict[str, Any]]

Choices the run put to the user and the answers it got. Recorded so the report can say a result rests on a decision somebody made, which is otherwise invisible in the numbers.

ran(tool: str) → bool

True when tool has already completed successfully.

reopen(names: Iterable[str]) → None

Let names run again: their earlier attempts no longer count.

For the steps downstream of one that was run again with other settings: what they read has changed, so what they made is out of date.

Examples

>>> ctx = RunContext('x')
>>> _ = ctx.begin('evaluate'); ctx.finish()
>>> ctx.attempted('evaluate')
1
>>> ctx.reopen(['evaluate'])
>>> ctx.attempted('evaluate')
0

reopened: Dict[str, int]

Tool name -> index of the first step that counts as an attempt of it. A step run again with other settings changes what the steps after it read; those are reopened (reopen()), so they can run again on the new result instead of being refused as already tried.

results() → Dict[str, Any]

The artifacts, as the dictionary the existing callers expect.

settings: Dict[str, Any]

steps: List[Step]

transcript(limit: int = 40) → str

The run so far, as the controller reads it.

Parameters:

limit (int) – Most recent steps to include. A run that loops is the one that most needs its recent history; the early steps are represented by the artifacts they produced.

Returns:

Goal, artifacts on hand, and the step log.

Return type:

str

unfinished() → List[str]

Tools that were attempted and did not succeed.

warnings: List[str]

class PyHydroGeophysX.agents.runtime.Step(tool: str, reason: str = '', status: str = 'ok', summary: str = '', error: str = '', produced: List[str] = <factory>, seconds: float = 0.0, agent: str = '', description: str = '', purpose: str = '')

Bases: object

One tool invocation and what came of it.

agent: str = ''

description: str = ''

error: str = ''

line() → str

The step as one line of transcript.

produced: List[str]

purpose: str = ''

What the tool is for, in its own words. Distinct from reason, which is why this run chose it; without a model to ask there is no reason, and the plan then read “next available step” for every row.

reason: str = ''

Why this step was chosen, in the controller’s words. Recorded because a reader of the finished report asks “why did it do that”, and the answer is otherwise nowhere.

seconds: float = 0.0

status: str = 'ok'

summary: str = ''

One or two sentences of what the tool concluded - not its output data.

tool: str

class PyHydroGeophysX.agents.runtime.Tool(name: str, description: str, handler: ~typing.Callable[[~PyHydroGeophysX.agents.runtime.context.RunContext], ~typing.Tuple[str, ~typing.Dict[str, ~typing.Any]]], requires: ~typing.Sequence[str] = <factory>, produces: ~typing.Sequence[str] = <factory>, agent: str = '', label: str = '', module: str = '', when: ~typing.Callable[[~PyHydroGeophysX.agents.runtime.context.RunContext], bool] | None = None, repeatable: bool = False, proposable: bool = True, applies: ~typing.Callable[[~PyHydroGeophysX.agents.runtime.context.RunContext], bool] | None = None)

Bases: object

One capability the controller can choose.

agent: str = ''

Which agent class implements it, for the derived execution plan.

applies: Callable[[RunContext], bool] | None = None

whether the tool applies to this run at all, whatever its gate says (a time-lapse inversion needs two surveys).

Type:

For an off-route proposal

available(ctx: RunContext) → bool

Whether this tool can run against the context as it stands.

description: str

handler: Callable[[RunContext], Tuple[str, Dict[str, Any]]]

label: str = ''

A short imperative label for the plan, e.g. “Run time-lapse inversion”.

missing(ctx: RunContext) → List[str]

Requirements that are not satisfied, for a blocked-step message.

module: str = ''

The desktop studio module that shows this kind of work, as keyed in PyHydroGeophysX.qt_apps.modules.MODULE_SPECS. The runtime never imports Qt; it reports the key and the desktop decides what to do with it, which is what lets a run walk the user through the studio while it works instead of leaving them on a progress bar.

name: str

offer() → str

The tool as one line of the controller’s menu.

produces: Sequence[str]

proposable: bool = True

Whether the controller may propose this tool off the standard route (toolbox) when its when gate is closed. Not for a loader, which reads a file the configuration has to name.

repeatable: bool = False

Tools that make no sense to run twice. The controller will not re-offer one that has already been attempted unless it is marked repeatable.

requires: Sequence[str]

unavailable_because(ctx: RunContext) → str

Why available() is False, in words the controller can act on.

when: Callable[[RunContext], bool] | None = None

Extra condition beyond requires - for tools gated by configuration rather than by data, such as the climate step.

PyHydroGeophysX.agents.runtime.announcement(ctx: RunContext, tool: Any, reason: str) → Dict[str, Any]

What the app is told about the step that is about to run.

Parameters:

• ctx (RunContext) – The run as it stands.

• tool (Tool or None) – The tool about to run. None when the controller named something that is not registered, which is itself worth announcing.

• reason (str) – Why the controller chose it.

Returns:

{"tool", "label", "module", "reason", "step"} - all JSON-safe. module is a studio module key or "".

Return type:

dict

Raises:

None –

Examples

>>> from .tools import Tool
>>> ctx = RunContext('goal')
>>> tool = Tool('invert_ert', 'Invert.', lambda c: ('', {}),
...             label='Run ERT inversion', module='ert')
>>> announcement(ctx, tool, 'the data are loaded') == {
...     'tool': 'invert_ert', 'label': 'Run ERT inversion', 'module': 'ert',
...     'reason': 'the data are loaded', 'step': 1}
True
>>> announcement(ctx, None, '')['label']
'Unknown step'

PyHydroGeophysX.agents.runtime.auto(announce: Callable[[Dict[str, Any]], None] | None = None) → Callable[[...], str]

An on_step hook that runs straight through, narrating as it goes.

Parameters:

announce (callable, optional) – Receives the announcement() for each step before it runs. The desktop uses it to bring the relevant module to the front, so the user watches the work happen rather than a progress bar.

Returns:

Suitable as run_controller(..., on_step=...). Always proceeds; an announcement that raises is ignored, because a display problem must not stop a computation.

Return type:

callable

Raises:

None –

Examples

>>> from .tools import Tool
>>> seen = []
>>> hook = auto(seen.append)
>>> tool = Tool('invert_ert', 'Invert.', lambda c: ('', {}), module='ert')
>>> hook(RunContext('goal'), tool, 'because')
'proceed'
>>> seen[0]['module']
'ert'
>>> auto(lambda event: 1 / 0)(RunContext('g'), tool, '')
'proceed'

PyHydroGeophysX.agents.runtime.completion(ctx: RunContext, step: Any, tools: Dict[str, Any] | None = None) → Dict[str, Any]

What the app is told about a step that has just finished.

The counterpart to announcement(). That one fires before a step and so can only say what is about to happen; this carries what the step actually concluded, which is the only evidence of work a user can see while a run is still going - the studio’s own panels stay empty, because the workflow runs headless in its own process.

Returns:

{"tool", "label", "module", "summary", "status", "produced", "step"} - all JSON-safe, since this crosses a process boundary as one line.

Return type:

dict

Examples

>>> ctx = RunContext('goal')
>>> ctx.begin('load_ert_surveys', 'it is first',
...           description='Load ERT data')
...
Step(tool='load_ert_surveys', ...)
>>> ctx.put('ert_data', [1, 2])
>>> ctx.finish('Loaded 5 surveys, 812 measurements.')
>>> event = completion(ctx, ctx.steps[-1])
>>> event['summary'], event['status'], event['produced']
('Loaded 5 surveys, 812 measurements.', 'ok', ['ert_data'])

PyHydroGeophysX.agents.runtime.register(tool: Tool) → Tool

Add tool to the registry, replacing any tool of the same name.

Parameters:

tool (Tool) – The tool to register.

Returns:

The same tool, so this can wrap a definition.

Return type:

Tool

Raises:

ValueError – If the tool has no name or no handler - a nameless tool cannot be chosen and a handlerless one cannot be run, and both fail far from here.

Examples

>>> t = register(Tool('noop', 'Do nothing.', lambda ctx: ('done', {})))
>>> TOOLS['noop'] is t
True
>>> del TOOLS['noop']

PyHydroGeophysX.agents.runtime.run_controller(ctx: RunContext, ask: Callable[[str], str] | None = None, max_steps: int = 24, progress: Callable[[str, float, str], None] | None = None, tools: Dict[str, Any] | None = None, on_step: Callable[[...], str | None] | None = None, on_result: Callable[[...], None] | None = None, recovery: bool = True, prompt: str | None = None, finish: str = 'report_files', flexible: bool = False, approves_steps: bool = False) → RunContext

Drive the run to completion, one chosen action at a time.

Parameters:

• ctx (RunContext) – The run. Mutated in place and returned.

• ask (callable, optional) – Takes a prompt, returns the model’s reply. Omit to run the deterministic policy - which is what happens without an API key.

• max_steps (int) – Hard bound on iterations, so a model that keeps choosing badly still terminates with whatever it has produced.

• progress (callable, optional) – (step, fraction, detail), called before each action so the desktop app can show the run advancing.

• on_step (callable, optional) – (ctx, tool, reason) -> 'proceed' | 'skip' | 'stop', called before each action with the tool about to run. This is the single difference between the app’s two modes: “auto to report” supplies a hook that notes the tool’s module and returns immediately, while “step by step” supplies one that shows the user what is about to happen and waits. Both drive the same loop over the same tools, so a step behaves identically whichever way it was reached - the previous design had two separate code paths that could and did drift.

• recovery (bool) – Whether a failed step may be diagnosed and retried once with different settings. On by default, because the alternative - recording the failure and moving on - loses runs that one changed number would have saved. Only configuration is ever changed, never code and never a file path; see recovery.

• on_result (callable, optional) – (ctx, step), called after each action with the step it finished. on_step fires before a step and so can only say what is about to happen; this is where what a step actually concluded becomes available to a caller while the run is still going.

• prompt (str, optional) – The decision prompt, with {transcript} and {menu} fields. Defaults to CONTROLLER_PROMPT; an assistant for another domain passes its own framing.

• finish (str) – The artifact that must exist before “done” is accepted (DEFAULT_FINISH, the report, unless the assistant says otherwise).

• flexible (bool) – Let the model step off the standard route (toolbox): look at a product’s numbers, add a step the route has not scheduled, run a finished step again with other settings, change a setting, leave route steps out, write a temporary script, or ask the user. Everything but a look is put to the user first with the model’s reason (toolbox.propose()), and the route is the default. The model is asked with FLEXIBLE_PROMPT, and says in its answer which steps it means to run next and which results it wants to see; those steps then run without asking it again (_follow_plan()) until it wants to look, or something it could not have planned for happens. Ignored without ask, and with a prompt of the assistant’s own, which does not offer it.

• approves_steps (bool) – on_step asks the user before every step (step by step), so an off-route step is approved there, with its reason, rather than in a question of its own.

Returns:

The same context, carrying the steps taken, the artifacts produced, any warnings raised and, in ended, how the loop ended. A model’s “done” ends it only once a report exists; before that the next available step is taken instead.

Return type:

RunContext

Raises:

None – Tool failures are recorded, not raised: the point of the loop is to keep deciding after one.

Examples

>>> from .tools import Tool
>>> only = {'load': Tool('load', 'Load.',
...                      lambda c: ('Loaded 3 files.', {'data': [1, 2, 3]}),
...                      produces=('data',)),
...         'report': Tool('report', 'Report.', lambda c: ('Wrote it.', {}),
...                        requires=('data',))}
>>> ctx = run_controller(RunContext('process my data'), tools=only)
>>> [s.tool for s in ctx.steps]
['load', 'report']
>>> ctx.steps[0].summary
'Loaded 3 files.'
>>> ctx.ended
'exhausted'

PyHydroGeophysX.agents.runtime.runnable_tools(ctx: RunContext, tools: Dict[str, Tool] | None = None) → List[Tool]

Every tool whose requirements the context currently satisfies.

Parameters:

ctx (RunContext) – The run as it stands.

Returns:

In registration order, which is roughly the order a run needs them - so a controller that simply takes the first is already a sane pipeline.

Return type:

list of Tool

Examples

>>> ctx = RunContext('x')
>>> only = {'needs_data': Tool('needs_data', 'Needs data.',
...                            lambda c: ('', {}), requires=('data',))}
>>> [t.name for t in runnable_tools(ctx, only)]
[]
>>> ctx.put('data', [1])
>>> [t.name for t in runnable_tools(ctx, only)]
['needs_data']

PyHydroGeophysX.agents.runtime.step_by_step(approve: Callable[[Dict[str, Any]], Any], announce: Callable[[Dict[str, Any]], None] | None = None) → Callable[[...], str]

An on_step hook that asks before each step and honours the answer.

Parameters:

• approve (callable) – Receives the announcement() and returns what to do: True or 'proceed' to run it, 'skip' to leave it out, 'stop' or False to end the run here. Blocking is expected - in the desktop this is where the user reads the step and presses a button.

• announce (callable, optional) – Called before approve, with the same event, so the app can navigate to the module first and let the user see the panel they are deciding about.

Returns:

Suitable as run_controller(..., on_step=...).

Return type:

callable

Raises:

None – An approve that raises is treated as “stop”: a run whose approval gate is broken must not carry on unapproved.

Examples

>>> from .tools import Tool
>>> tool = Tool('invert_ert', 'Invert.', lambda c: ('', {}), module='ert')
>>> ctx = RunContext('goal')
>>> step_by_step(lambda event: True)(ctx, tool, '')
'proceed'
>>> step_by_step(lambda event: 'skip')(ctx, tool, '')
'skip'
>>> step_by_step(lambda event: False)(ctx, tool, '')
'stop'
>>> step_by_step(lambda event: 1 / 0)(ctx, tool, '')
'stop'