diff --git a/README.md b/README.md index d86fbb0..93e0c83 100644 --- a/README.md +++ b/README.md @@ -797,6 +797,13 @@ Two-stage flow: optional `.set` parameter file. Returns a `jobId`; poll for status; fetch the report and terminal log when complete. +`POST /backtest` also accepts an auditable project mode used by `mt5-cli`: +upload `ini`, `project_manifest`, and `project_bundle` together. In that mode +the API validates every declared path, size, SHA-256, and ZIP entry, replaces +the terminal's project modules only while the job owns the run lock, captures +outputs and log deltas, and restores the exact pre-run trees before publishing +the final status. Project mode never falls back to the ad-hoc expert/set flow. + For a plain backtest, leave `[Tester].Optimization=0` or omit it. For an optimization, set `[Tester].Optimization` to one of: @@ -1039,6 +1046,36 @@ Multipart form fields: | `topPasses` | no | For optimization jobs, keep the top `1..500` parsed XML passes in the status payload. Default `50`. | | `timeout` | no | Duration string override (`"30m"`, `"6h"`, `"3h30m"`). Defaults to `backtest_timeout` from `config.yaml`, then hardcoded `6h`. | +Project-mode multipart fields are a separate, complete contract: + +| Field | Required | Notes | +| ------------------ | -------- | ----- | +| `ini` | yes | UTF-8 runtime INI. `[Common]` credentials are replaced by this API's configured account before launch. | +| `project_manifest` | yes | JSON schema `MT5-CLI-PROJECT-MATERIALIZATION-001` with `projectId`, external `runId`, `planFingerprint`, exact Expert/preset paths, and the complete artifact inventory. | +| `project_bundle` | yes | ZIP containing exactly one byte payload for every manifest artifact. Extra, missing, duplicate, unsafe, or hash-divergent entries are rejected. | +| `timeout` | no | Same real duration override used by the ad-hoc mode. | + +Each artifact declares `module`, project-relative `relativePath`, unique +`bundleEntry`, `size`, and `sha256`. Supported modules and their physical +remote destinations are: + +| Module | Remote destination | +| ------ | ------------------ | +| `EXP` | `MQL5/Experts/` | +| `IND` | `MQL5/Indicators/` | +| `SCR` | `MQL5/Scripts/` | +| `SRV` | `MQL5/Services/` | +| `INC` | `MQL5/Include/` | +| `LBR` | `MQL5/Libraries/` | +| `PRS` | `MQL5/Presets/` | +| `TPL` | `Profiles/Templates/` | +| `FLS` | terminal Common `Files/` | + +Only one project directory per module is accepted. Physical existing module +trees are backed up and compared byte for byte after restoration. Reparse +points, symlinks, junctions, physical conflicts that cannot be restored, and +undeclared files are explicit failures. + Responds `202 Accepted` with `Retry-After` header and the queued job payload: ```json @@ -1064,6 +1101,29 @@ overwritten with the credentials from `config.yaml` for the request's broker/account. The expert path is rewritten to `Uploaded\` and the set file is namespaced per job to avoid collisions. +In project mode, the Expert and preset paths are not rewritten to `Uploaded`. +They must match `expertRelativePath` and `presetRelativePath` from the validated +project manifest. The submitted credential-free INI remains available as raw +evidence, while the normalized launch INI is generated inside the job stage. + +Project-mode responses use `pollAfterSeconds: 2` and expose the exact evidence +URLs below in addition to the normal status/report/log fields: + +- `submittedIniUrl` +- `projectManifestUrl` +- `projectBundleUrl` +- `executionManifestUrl` +- `materializationAuditUrl` +- `outputManifestUrl` +- `outputArtifactsUrl` +- `logDeltaManifestUrl` +- `logDeltaArtifactsUrl` +- `cancelUrl` + +The output archive contains only created or modified `FLS` files. Its manifest +preserves every path, classification, size, SHA-256, and ZIP entry; unchanged +files remain explicitly classified rather than copied as new output. + #### `GET /backtest/` Status payload. `status` ∈ `queued` `running` `completed` `failed`. When @@ -1162,6 +1222,34 @@ Stream the raw report and terminal log file. Backtests return the MT5 HTML report. Optimizations return the MT5 XML spreadsheet export. `404` until the job finishes. +#### Project evidence and cancellation endpoints + +For a project-mode job, raw evidence is available through: + +```text +GET /backtest//submitted-ini +GET /backtest//project-manifest +GET /backtest//project-bundle +GET /backtest//execution-manifest +GET /backtest//materialization +GET /backtest//output-manifest +GET /backtest//artifacts +GET /backtest//log-deltas-manifest +GET /backtest//log-deltas +GET /backtest//tail?lines=200 +DELETE /backtest/ +``` + +`DELETE` is accepted only for project-mode ownership and targets only that +job's recorded subprocess. It sets `cancelRequested`, attempts termination of +the exact owned process, and records every termination action. It never kills a +terminal by executable name or adopts another job. + +The project worker always attempts output capture, log-delta capture, project +restoration, and report restoration before final status. Any failure in those +phases is preserved in the execution manifest and prevents a false successful +completion. + #### Worked example ```bash diff --git a/mt5api/backtest/handler.py b/mt5api/backtest/handler.py index 5d714a6..afc6d3f 100644 --- a/mt5api/backtest/handler.py +++ b/mt5api/backtest/handler.py @@ -14,7 +14,9 @@ """ from __future__ import annotations +import base64 import configparser +import hashlib import io import json import os @@ -23,13 +25,22 @@ import threading import time import uuid +import zipfile from flask import Response, abort, jsonify, request, send_file -from mt5api.backtest import cache_parser, ini_builder, jobs, optimization_parser, set_builder +from mt5api.backtest import ( + cache_parser, + ini_builder, + jobs, + materialization, + optimization_parser, + set_builder, +) from mt5api.config import ( ACCOUNT, BROKER, + COMMON_FILES_DIR, LOG_DIR, TERMINAL_DIR, TERMINAL_PATH, @@ -44,9 +55,16 @@ from mt5api.logger import log RUN_LOCK = threading.Lock() +ACTIVE_PROCESS_LOCK = threading.Lock() +ACTIVE_PROCESSES = {} DIAGNOSTIC_TAIL_CHARS = 4000 DEFAULT_TOP_PASSES = 50 MAX_TOP_PASSES = 500 +OWNED_PROCESS_TERMINATE_SECONDS = 15 + + +class ProjectJobCanceled(Exception): + pass # ── INI builder route ─────────────────────────────────────────────── @@ -278,6 +296,54 @@ def _parse_top_passes(raw_value): return value +def _save_upload(upload, path): + os.makedirs(os.path.dirname(path), exist_ok=True) + upload.save(path) + + +def _canonical_ini_path(value, *, remove_ex5=False): + canonical = (value or "").strip().replace("/", "\\") + while "\\\\" in canonical: + canonical = canonical.replace("\\\\", "\\") + if remove_ex5 and canonical.lower().endswith(".ex5"): + canonical = canonical[:-4] + return canonical + + +def _validate_project_ini_contract(parser, project): + tester = parser["Tester"] + actual_expert = _canonical_ini_path(tester.get("Expert", ""), remove_ex5=True) + required_expert = _canonical_ini_path(project.expert_ini_value, remove_ex5=True) + if actual_expert.casefold() != required_expert.casefold(): + raise materialization.MaterializationError( + "ini_expert_manifest_mismatch", + "INI [Tester].Expert does not identify expertRelativePath from the project manifest", + details={"actual": actual_expert, "required": required_expert}, + ) + actual_preset = _canonical_ini_path(tester.get("ExpertParameters", "")) + required_preset = _canonical_ini_path(project.preset_ini_value) + if actual_preset.casefold() != required_preset.casefold(): + raise materialization.MaterializationError( + "ini_preset_manifest_mismatch", + "INI [Tester].ExpertParameters does not identify presetRelativePath from the project manifest", + details={"actual": actual_preset, "required": required_preset}, + ) + tester["Expert"] = required_expert + if required_preset: + tester["ExpertParameters"] = required_preset + else: + tester.pop("ExpertParameters", None) + + +def _write_json_evidence(path, payload): + os.makedirs(os.path.dirname(path), exist_ok=True) + temporary = f"{path}.tmp" + with open(temporary, "w", encoding="utf-8", newline="\n") as handle: + json.dump(payload, handle, indent=2, sort_keys=True, ensure_ascii=False) + handle.write("\n") + os.replace(temporary, path) + + # ── Submit route ──────────────────────────────────────────────────── @@ -294,6 +360,39 @@ def run_backtest(): except UnicodeDecodeError: return jsonify({"error": "INI must be UTF-8 text"}), 400 + project_manifest_upload = request.files.get("project_manifest") + project_bundle_upload = request.files.get("project_bundle") + project_mode = bool( + (project_manifest_upload is not None and project_manifest_upload.filename) + or (project_bundle_upload is not None and project_bundle_upload.filename) + ) + if project_mode and not ( + project_manifest_upload is not None + and project_manifest_upload.filename + and project_bundle_upload is not None + and project_bundle_upload.filename + ): + return jsonify({ + "error": "project_manifest and project_bundle are both required for project mode", + "projectManifestPresent": bool( + project_manifest_upload is not None and project_manifest_upload.filename + ), + "projectBundlePresent": bool( + project_bundle_upload is not None and project_bundle_upload.filename + ), + }), 400 + + job_id = uuid.uuid4().hex + stage_dir = os.path.join(BACKTEST_JOB_DIR, job_id) + os.makedirs(stage_dir, exist_ok=True) + os.makedirs(LOG_DIR, exist_ok=True) + submitted_ini_path = os.path.join(stage_dir, "submitted.ini") + with open(submitted_ini_path, "w", encoding="utf-8", newline="\n") as handle: + handle.write(ini_text) + project_manifest_path = os.path.join(stage_dir, "project-manifest.json") + project_bundle_path = os.path.join(stage_dir, "project-bundle.zip") + project = None + try: timeout_value = (request.form.get("timeout") or "").strip() top_passes = _parse_top_passes(request.form.get("topPasses")) @@ -302,57 +401,143 @@ def run_backtest(): if timeout_value else BACKTEST_TIMEOUT_SECONDS ) - expert_filename, expert_bytes = _read_submission( - request.files.get("expert"), - request.form.get("expert_name", ""), - "experts", - "expert", - required=True, - required_ext=".ex5", - ) - set_filename, set_bytes = _read_submission( - request.files.get("set"), - request.form.get("set_name", ""), - "sets", - "set", - required=False, - required_ext=".set", - ) parser = _parse_ini(ini_text) creds = _load_account_config() _override_credentials(parser, creds) _normalize_symbol(parser) - _normalize_expert(parser, expert_filename) + if project_mode: + _save_upload(project_manifest_upload, project_manifest_path) + _save_upload(project_bundle_upload, project_bundle_path) + project = materialization.ProjectMaterialization( + project_manifest_path, + project_bundle_path, + stage_dir, + TERMINAL_DIR, + COMMON_FILES_DIR, + ) + _validate_project_ini_contract(parser, project) + expert_filename = os.path.basename(project.manifest.expert_relative_path) + set_filename = ( + os.path.basename(project.manifest.preset_relative_path) + if project.manifest.preset_relative_path + else "" + ) + expert_bytes = b"" + set_bytes = b"" + else: + expert_filename, expert_bytes = _read_submission( + request.files.get("expert"), + request.form.get("expert_name", ""), + "experts", + "expert", + required=True, + required_ext=".ex5", + ) + set_filename, set_bytes = _read_submission( + request.files.get("set"), + request.form.get("set_name", ""), + "sets", + "set", + required=False, + required_ext=".set", + ) + _normalize_expert(parser, expert_filename) optimization_type = _optimization_type(parser) report_name = _ensure_report_path(parser) - except ValueError as exc: + except (ValueError, OSError) as exc: + if project_mode: + error = ( + exc.as_dict() + if isinstance(exc, materialization.MaterializationError) + else {"code": type(exc).__name__, "message": str(exc)} + ) + submission_error_path = os.path.join(stage_dir, "submission-error.json") + _write_json_evidence(submission_error_path, { + "jobId": job_id, + "status": "failed", + "failedAt": jobs.now_iso(), + "error": error, + "projectManifestPath": ( + project_manifest_path if os.path.exists(project_manifest_path) else None + ), + "projectBundlePath": ( + project_bundle_path if os.path.exists(project_bundle_path) else None + ), + "submittedIniPath": submitted_ini_path, + }) + failed_job = { + "jobId": job_id, + "status": "failed", + "broker": BROKER, + "account": ACCOUNT, + "submittedAt": jobs.now_iso(), + "startedAt": None, + "finishedAt": jobs.now_iso(), + "durationSeconds": None, + "reportName": None, + "reportPath": None, + "logPath": None, + "stageDir": stage_dir, + "submittedIniPath": submitted_ini_path, + "projectMode": True, + "projectId": project.manifest.project_id if project else None, + "externalRunId": project.manifest.run_id if project else None, + "planFingerprint": project.manifest.plan_fingerprint if project else None, + "projectManifestPath": ( + project_manifest_path if os.path.exists(project_manifest_path) else None + ), + "projectBundlePath": ( + project_bundle_path if os.path.exists(project_bundle_path) else None + ), + "materializationAuditPath": ( + project.audit_path + if project and os.path.exists(project.audit_path) + else None + ), + "submissionErrorPath": submission_error_path, + "materializationStatus": "validation_failed", + "cancelRequested": False, + "exitCode": None, + "error": error, + "summary": None, + "optimizationType": 0, + "optimizationResults": None, + "optimizationCache": None, + } + jobs.store_job(failed_job) + return jsonify({**jobs.public_payload(failed_job), "error": error}), 400 return jsonify({"error": str(exc)}), 400 - job_id = uuid.uuid4().hex # Namespace the .set so concurrent jobs cannot clobber the same file in # MQL5\Profiles\Tester\. - staged_set_filename = f"{job_id}__{set_filename}" if set_filename else "" - _normalize_set(parser, staged_set_filename) - - stage_dir = os.path.join(BACKTEST_JOB_DIR, job_id) - os.makedirs(stage_dir, exist_ok=True) - os.makedirs(LOG_DIR, exist_ok=True) + staged_set_filename = ( + "" + if project_mode + else (f"{job_id}__{set_filename}" if set_filename else "") + ) + if not project_mode: + _normalize_set(parser, staged_set_filename) - staged_expert_path = os.path.join(stage_dir, expert_filename) - with open(staged_expert_path, "wb") as handle: - handle.write(expert_bytes) + staged_expert_path = "" staged_set_path = "" - if set_filename: - staged_set_path = os.path.join(stage_dir, staged_set_filename) - with open(staged_set_path, "wb") as handle: - handle.write(set_bytes) + if not project_mode: + staged_expert_path = os.path.join(stage_dir, expert_filename) + with open(staged_expert_path, "wb") as handle: + handle.write(expert_bytes) + if set_filename: + staged_set_path = os.path.join(stage_dir, staged_set_filename) + with open(staged_set_path, "wb") as handle: + handle.write(set_bytes) # Save the human-readable normalized INI for debugging. debug_ini_path = os.path.join(stage_dir, "normalized.ini") with open(debug_ini_path, "w", encoding="utf-8", newline="\n") as handle: handle.write(_serialize_ini(parser)) - report_path = os.path.join(TERMINAL_DIR, "Reports", report_name) + terminal_report_path = os.path.join(TERMINAL_DIR, "Reports", report_name) + report_path = ( + os.path.join(stage_dir, report_name) if project_mode else terminal_report_path + ) log_path = os.path.join(stage_dir, "run.log") job = { @@ -366,8 +551,10 @@ def run_backtest(): "durationSeconds": None, "reportName": report_name, "reportPath": report_path, + "terminalReportPath": terminal_report_path, "logPath": log_path, "debugIniPath": debug_ini_path, + "submittedIniPath": submitted_ini_path, "stageDir": stage_dir, "expertFilename": expert_filename, "setFilename": set_filename, @@ -382,6 +569,29 @@ def run_backtest(): "optimizationCache": None, "topPasses": top_passes, "timeoutSeconds": timeout_seconds, + "projectMode": project_mode, + "projectId": project.manifest.project_id if project else None, + "externalRunId": project.manifest.run_id if project else None, + "planFingerprint": project.manifest.plan_fingerprint if project else None, + "projectManifestPath": project_manifest_path if project_mode else None, + "projectBundlePath": project_bundle_path if project_mode else None, + "materializationAuditPath": project.audit_path if project else None, + "outputManifestPath": project.output_manifest_path if project else None, + "outputArtifactsPath": project.output_archive_path if project else None, + "executionManifestPath": ( + os.path.join(stage_dir, "execution-manifest.json") if project_mode else None + ), + "logDeltaManifestPath": ( + os.path.join(stage_dir, "log-deltas-manifest.json") if project_mode else None + ), + "logBaselinePath": ( + os.path.join(stage_dir, "log-baseline.json") if project_mode else None + ), + "logDeltaArtifactsPath": ( + os.path.join(stage_dir, "log-deltas.zip") if project_mode else None + ), + "materializationStatus": "validated" if project_mode else None, + "cancelRequested": False, } jobs.store_job(job) @@ -389,7 +599,9 @@ def run_backtest(): response = jsonify(jobs.public_payload(job)) response.status_code = 202 - response.headers["Retry-After"] = str(jobs.POLL_AFTER_SECONDS) + response.headers["Retry-After"] = str( + jobs.PROJECT_POLL_AFTER_SECONDS if project_mode else jobs.POLL_AFTER_SECONDS + ) return response @@ -397,6 +609,545 @@ def run_backtest(): def _execute_job(job_id): + job = jobs.load_job(job_id) + if job is None: + return + if job.get("projectMode"): + _execute_project_job(job_id) + return + _execute_legacy_job(job_id) + + +def _log_inventory(): + inventory = {} + for scope, log_dir in ( + ("terminal", os.path.join(TERMINAL_DIR, "logs")), + ("tester", os.path.join(TERMINAL_DIR, "Tester", "logs")), + ): + if not os.path.isdir(log_dir): + continue + for current_root, _, file_names in os.walk(log_dir): + for file_name in sorted(file_names): + if not file_name.lower().endswith(".log"): + continue + path = os.path.join(current_root, file_name) + if not os.path.isfile(path): + continue + relative = os.path.relpath(path, log_dir).replace(os.sep, "/") + inventory[f"{scope}/{relative}"] = { + "scope": scope, + "relativePath": relative, + "path": path, + "size": os.path.getsize(path), + "sha256": materialization.sha256_file(path), + } + return dict(sorted(inventory.items(), key=lambda item: item[0].casefold())) + + +def _hash_prefix(path, length): + digest = hashlib.sha256() + remaining = length + with open(path, "rb") as handle: + while remaining > 0: + chunk = handle.read(min(1024 * 1024, remaining)) + if not chunk: + break + digest.update(chunk) + remaining -= len(chunk) + return digest.hexdigest(), length - remaining + + +def _copy_file_range(source_path, destination_path, offset): + os.makedirs(os.path.dirname(destination_path), exist_ok=True) + with open(source_path, "rb") as source, open(destination_path, "wb") as destination: + source.seek(offset) + shutil.copyfileobj(source, destination, length=1024 * 1024) + + +def _capture_log_deltas(job, baseline): + current = _log_inventory() + records = [] + delta_root = os.path.join(job["stageDir"], "log-deltas") + for key in sorted(set(baseline) | set(current), key=lambda value: value.casefold()): + before = baseline.get(key) + after = current.get(key) + classification = None + offset = 0 + if before is None: + classification = "created" + elif after is None: + classification = "missing_after_run" + elif before["size"] == after["size"] and before["sha256"] == after["sha256"]: + classification = "unchanged" + elif after["size"] >= before["size"]: + prefix_sha256, prefix_size = _hash_prefix(after["path"], before["size"]) + if prefix_size == before["size"] and prefix_sha256 == before["sha256"]: + classification = "appended" + offset = before["size"] + else: + classification = "replaced" + else: + classification = "truncated_or_replaced" + + delta_path = None + delta_size = None + delta_sha256 = None + if after is not None and classification not in ("unchanged", "missing_after_run"): + delta_path = os.path.join(delta_root, *key.split("/")) + _copy_file_range(after["path"], delta_path, offset) + delta_size = os.path.getsize(delta_path) + delta_sha256 = materialization.sha256_file(delta_path) + archive_entry = f"deltas/{key}" if delta_path else None + records.append({ + "key": key, + "classification": classification, + "before": before, + "after": after, + "capturedOffset": offset if delta_path else None, + "deltaPath": delta_path, + "deltaSize": delta_size, + "deltaSha256": delta_sha256, + "archiveEntry": archive_entry, + }) + with zipfile.ZipFile( + job["logDeltaArtifactsPath"], + "w", + compression=zipfile.ZIP_DEFLATED, + allowZip64=True, + ) as archive: + for record in records: + if record["deltaPath"]: + archive.write(record["deltaPath"], arcname=record["archiveEntry"]) + payload = { + "schemaVersion": "MT5-CLI-REMOTE-LOG-DELTAS-001", + "jobId": job["jobId"], + "projectId": job.get("projectId"), + "externalRunId": job.get("externalRunId"), + "capturedAt": jobs.now_iso(), + "archivePath": job["logDeltaArtifactsPath"], + "archiveSize": os.path.getsize(job["logDeltaArtifactsPath"]), + "archiveSha256": materialization.sha256_file(job["logDeltaArtifactsPath"]), + "files": records, + } + _write_json_evidence(job["logDeltaManifestPath"], payload) + return payload + + +def _prepare_report_backup(job): + paths = [job["terminalReportPath"]] + symbols_path = f"{os.path.splitext(job['terminalReportPath'])[0]}.symbols.xml" + if os.path.normcase(symbols_path) != os.path.normcase(paths[0]): + paths.append(symbols_path) + evidence = [] + backup_root = os.path.join(job["stageDir"], "report-backups") + for index, path in enumerate(paths): + present = os.path.isfile(path) + backup_path = os.path.join(backup_root, f"{index:02d}-{os.path.basename(path)}") + record = { + "path": path, + "present": present, + "size": os.path.getsize(path) if present else None, + "sha256": materialization.sha256_file(path) if present else None, + "backupPath": backup_path if present else None, + } + if present: + os.makedirs(os.path.dirname(backup_path), exist_ok=True) + shutil.copyfile(path, backup_path) + if materialization.sha256_file(backup_path) != record["sha256"]: + raise materialization.MaterializationError( + "report_backup_integrity_mismatch", + "Existing terminal report backup failed SHA-256 verification", + details=record, + ) + if os.path.exists(path): + os.remove(path) + evidence.append(record) + return evidence + + +def _collect_project_report(job): + source = job["terminalReportPath"] + report_name = job["reportName"] + if not os.path.isfile(source) and job.get("optimizationType") == 3: + symbols_path = f"{os.path.splitext(source)[0]}.symbols.xml" + if os.path.isfile(symbols_path): + source = symbols_path + report_name = os.path.basename(symbols_path) + if not os.path.isfile(source): + return None + destination = os.path.join(job["stageDir"], report_name) + shutil.copyfile(source, destination) + source_sha256 = materialization.sha256_file(source) + destination_sha256 = materialization.sha256_file(destination) + if source_sha256 != destination_sha256: + raise materialization.MaterializationError( + "report_capture_integrity_mismatch", + "Captured report differs from the terminal report", + details={ + "source": source, + "sourceSha256": source_sha256, + "destination": destination, + "destinationSha256": destination_sha256, + }, + ) + jobs.update_job(job["jobId"], reportPath=destination, reportName=report_name) + job["reportPath"] = destination + job["reportName"] = report_name + return { + "terminalPath": source, + "capturedPath": destination, + "size": os.path.getsize(destination), + "sha256": destination_sha256, + } + + +def _restore_report_backup(evidence): + failures = [] + for record in evidence: + try: + if os.path.exists(record["path"]): + os.remove(record["path"]) + if record["present"]: + os.makedirs(os.path.dirname(record["path"]), exist_ok=True) + shutil.copyfile(record["backupPath"], record["path"]) + restored_sha256 = materialization.sha256_file(record["path"]) + if restored_sha256 != record["sha256"]: + raise materialization.MaterializationError( + "report_restore_integrity_mismatch", + "Restored terminal report differs from its backup", + details={**record, "restoredSha256": restored_sha256}, + ) + record["restored"] = True + except Exception as exc: + error = exc.as_dict() if isinstance( + exc, materialization.MaterializationError + ) else {"code": type(exc).__name__, "message": str(exc)} + record["restored"] = False + record["restoreError"] = error + failures.append({"path": record["path"], **error}) + if failures: + raise materialization.MaterializationError( + "report_restore_failed", + "One or more terminal report paths could not be restored", + details=failures, + ) + + +def _terminate_owned_process(process): + actions = [] + if process.poll() is not None: + return actions + process.terminate() + actions.append({"action": "terminate", "at": jobs.now_iso()}) + try: + process.wait(timeout=OWNED_PROCESS_TERMINATE_SECONDS) + except subprocess.TimeoutExpired: + process.kill() + actions.append({"action": "kill", "at": jobs.now_iso()}) + process.wait(timeout=OWNED_PROCESS_TERMINATE_SECONDS) + return actions + + +def _execute_project_job(job_id): + job = jobs.load_job(job_id) + if job is None: + return + start_time = time.time() + execution = { + "schemaVersion": "MT5-CLI-REMOTE-EXECUTION-001", + "jobId": job_id, + "projectId": job.get("projectId"), + "externalRunId": job.get("externalRunId"), + "planFingerprint": job.get("planFingerprint"), + "broker": BROKER, + "account": ACCOUNT, + "submittedAt": job.get("submittedAt"), + "startedAt": None, + "finishedAt": None, + "command": None, + "cwd": TERMINAL_DIR, + "pid": None, + "exitCode": None, + "stdoutPath": job.get("logPath"), + "stderr": "combined_with_stdout", + "timeoutSeconds": job.get("timeoutSeconds"), + "timedOut": False, + "cancelRequested": False, + "terminationActions": [], + "reportBackups": [], + "report": None, + "logDeltaManifestPath": job.get("logDeltaManifestPath"), + "materializationAuditPath": job.get("materializationAuditPath"), + "outputManifestPath": job.get("outputManifestPath"), + "outputArtifactsPath": job.get("outputArtifactsPath"), + "errors": [], + } + _write_json_evidence(job["executionManifestPath"], execution) + project = None + project_applied = False + report_backups = [] + log_baseline = {} + log_baseline_captured = False + result_returncode = None + failure = None + canceled = False + + with RUN_LOCK: + current = jobs.load_job(job_id) or job + if current.get("cancelRequested"): + canceled = True + if canceled: + jobs.update_job( + job_id, + status="canceled", + finishedAt=jobs.now_iso(), + durationSeconds=round(time.time() - start_time, 3), + ) + execution["cancelRequested"] = True + execution["finishedAt"] = jobs.now_iso() + _write_json_evidence(job["executionManifestPath"], execution) + return + + started_at = jobs.now_iso() + execution["startedAt"] = started_at + jobs.update_job(job_id, status="running", startedAt=started_at) + try: + project = materialization.ProjectMaterialization( + job["projectManifestPath"], + job["projectBundlePath"], + job["stageDir"], + TERMINAL_DIR, + COMMON_FILES_DIR, + ) + jobs.update_job( + job_id, + materializationAuditPath=project.audit_path, + materializationStatus="validated", + ) + report_backups = _prepare_report_backup(job) + execution["reportBackups"] = report_backups + log_baseline = _log_inventory() + log_baseline_captured = True + _write_json_evidence(job["logBaselinePath"], { + "schemaVersion": "MT5-CLI-REMOTE-LOG-BASELINE-001", + "jobId": job_id, + "projectId": job.get("projectId"), + "externalRunId": job.get("externalRunId"), + "capturedAt": jobs.now_iso(), + "files": log_baseline, + }) + project.apply() + project_applied = True + jobs.update_job(job_id, materializationStatus="applied") + current = jobs.load_job(job_id) or job + if current.get("cancelRequested"): + raise ProjectJobCanceled("Project backtest was canceled before terminal launch") + + parser = _parse_ini(_read_text_best_effort(job["debugIniPath"])) + _validate_project_ini_contract(parser, project) + ini_path = os.path.join(job["stageDir"], "tester.ini") + _write_utf16_ini(parser, ini_path) + command = [TERMINAL_PATH, "/portable", f"/config:{ini_path}"] + execution["command"] = command + execution["iniPath"] = ini_path + execution["iniSize"] = os.path.getsize(ini_path) + execution["iniSha256"] = materialization.sha256_file(ini_path) + _write_json_evidence(job["executionManifestPath"], execution) + + log.info( + "project backtest start broker=%s account=%s job=%s external_run=%s project=%s report=%s", + BROKER, + ACCOUNT, + job_id, + job.get("externalRunId"), + job.get("projectId"), + job["reportName"], + ) + with open(job["logPath"], "w", encoding="utf-8") as log_handle: + process = subprocess.Popen( + command, + cwd=TERMINAL_DIR, + stdout=log_handle, + stderr=subprocess.STDOUT, + ) + execution["pid"] = process.pid + with ACTIVE_PROCESS_LOCK: + ACTIVE_PROCESSES[job_id] = process + _write_json_evidence(job["executionManifestPath"], execution) + try: + result_returncode = process.wait(timeout=job["timeoutSeconds"]) + except subprocess.TimeoutExpired: + execution["timedOut"] = True + execution["terminationActions"].extend(_terminate_owned_process(process)) + result_returncode = process.returncode + failure = f"Backtest timed out after {job['timeoutSeconds']}s" + finally: + with ACTIVE_PROCESS_LOCK: + if ACTIVE_PROCESSES.get(job_id) is process: + ACTIVE_PROCESSES.pop(job_id, None) + + current = jobs.load_job(job_id) or job + canceled = bool(current.get("cancelRequested")) + execution["cancelRequested"] = canceled + execution["terminationActions"].extend( + current.get("cancelTerminationActions") or [] + ) + execution["exitCode"] = result_returncode + execution["report"] = _collect_project_report(job) + except Exception as exc: + current = jobs.load_job(job_id) or job + canceled = bool(current.get("cancelRequested")) + if isinstance(exc, ProjectJobCanceled) and canceled: + execution["cancelRequested"] = True + else: + log.exception( + "project backtest crashed broker=%s account=%s job=%s", + BROKER, + ACCOUNT, + job_id, + ) + error = exc.as_dict() if isinstance( + exc, materialization.MaterializationError + ) else {"code": type(exc).__name__, "message": str(exc)} + execution["errors"].append({"phase": "execute", **error}) + failure = f"Project backtest crashed: {error['message']}" + finally: + if log_baseline_captured: + try: + _capture_log_deltas(job, log_baseline) + except Exception as exc: + error = exc.as_dict() if isinstance( + exc, materialization.MaterializationError + ) else {"code": type(exc).__name__, "message": str(exc)} + execution["errors"].append({"phase": "capture_logs", **error}) + failure = failure or f"Log delta capture failed: {error['message']}" + if project is not None and project_applied: + try: + project.capture_outputs() + jobs.update_job( + job_id, + outputManifestPath=project.output_manifest_path, + outputArtifactsPath=project.output_archive_path, + ) + except Exception as exc: + error = exc.as_dict() if isinstance( + exc, materialization.MaterializationError + ) else {"code": type(exc).__name__, "message": str(exc)} + execution["errors"].append({"phase": "capture_outputs", **error}) + failure = failure or f"Output capture failed: {error['message']}" + try: + project.restore() + jobs.update_job(job_id, materializationStatus="restored") + except Exception as exc: + error = exc.as_dict() if isinstance( + exc, materialization.MaterializationError + ) else {"code": type(exc).__name__, "message": str(exc)} + execution["errors"].append({"phase": "restore_materialization", **error}) + jobs.update_job(job_id, materializationStatus="restore_failed") + failure = failure or f"Project restoration failed: {error['message']}" + if report_backups: + try: + _restore_report_backup(report_backups) + except Exception as exc: + error = exc.as_dict() if isinstance( + exc, materialization.MaterializationError + ) else {"code": type(exc).__name__, "message": str(exc)} + execution["errors"].append({"phase": "restore_report", **error}) + failure = failure or f"Report restoration failed: {error['message']}" + + duration = round(time.time() - start_time, 3) + execution["finishedAt"] = jobs.now_iso() + execution["durationSeconds"] = duration + execution["exitCode"] = result_returncode + execution["reportBackups"] = report_backups + _write_json_evidence(job["executionManifestPath"], execution) + + if canceled: + jobs.update_job( + job_id, + status="canceled", + exitCode=result_returncode, + durationSeconds=duration, + finishedAt=jobs.now_iso(), + ) + return + if failure: + jobs.update_job( + job_id, + status="failed", + error=failure, + exitCode=result_returncode, + durationSeconds=duration, + finishedAt=jobs.now_iso(), + ) + return + if result_returncode != 0: + jobs.update_job( + job_id, + status="failed", + error=f"terminal64.exe exited with code {result_returncode}", + exitCode=result_returncode, + durationSeconds=duration, + finishedAt=jobs.now_iso(), + ) + return + if not job.get("reportPath") or not os.path.exists(job["reportPath"]): + jobs.update_job( + job_id, + status="failed", + error="Report not generated", + exitCode=result_returncode, + durationSeconds=duration, + finishedAt=jobs.now_iso(), + ) + return + + if job.get("optimizationType"): + top_passes = job.get("topPasses", DEFAULT_TOP_PASSES) + cache_details = cache_parser.parse_cache_details( + job["debugIniPath"], TERMINAL_DIR, top_passes + ) + optimization_results = list(cache_details.get("rows") or []) + optimization_cache = cache_details.get("cache") + if not optimization_results and job.get("optimizationType") != 3: + optimization_results = optimization_parser.parse_optimization_report( + job["reportPath"], top_passes + ) + jobs.update_job( + job_id, + status="completed", + exitCode=result_returncode, + durationSeconds=duration, + finishedAt=jobs.now_iso(), + optimizationResults=optimization_results, + optimizationCache=optimization_cache, + ) + return + + report_html = _read_text_best_effort(job["reportPath"]) + summary = jobs.parse_report_summary(report_html) + if jobs.is_empty_backtest_summary(summary): + jobs.update_job( + job_id, + status="failed", + error="Backtest produced empty report (Bars=0, Ticks=0, Symbols=0)", + exitCode=result_returncode, + durationSeconds=duration, + finishedAt=jobs.now_iso(), + summary=summary, + ) + return + jobs.update_job( + job_id, + status="completed", + exitCode=result_returncode, + durationSeconds=duration, + finishedAt=jobs.now_iso(), + summary=summary, + ) + + +def _execute_legacy_job(job_id): job = jobs.load_job(job_id) if job is None: return @@ -611,6 +1362,148 @@ def get_status(job_id): return jsonify(jobs.public_payload(job)) +def cancel_job(job_id): + job = jobs.load_job(job_id) + if job is None: + return jsonify({"error": f"Backtest job not found: {job_id}"}), 404 + if not job.get("projectMode"): + return jsonify({ + "error": "Cancellation ownership is available only for project-mode jobs", + "jobId": job_id, + "status": job.get("status"), + }), 409 + if job.get("status") in jobs.TERMINAL_STATUSES: + return jsonify(jobs.public_payload(job)) + + requested_action = {"action": "cancel_requested", "at": jobs.now_iso()} + jobs.update_job( + job_id, + cancelRequested=True, + status="canceling", + cancelTerminationActions=[requested_action], + ) + with ACTIVE_PROCESS_LOCK: + process = ACTIVE_PROCESSES.get(job_id) + actions = [requested_action] + if process is not None: + try: + actions = _terminate_owned_process(process) + except Exception as exc: + error = {"code": type(exc).__name__, "message": str(exc)} + updated = jobs.update_job( + job_id, + cancelRequested=True, + cancelError=error, + cancelTerminationActions=actions, + ) + return jsonify({**jobs.public_payload(updated), "cancelError": error}), 500 + updated = jobs.update_job( + job_id, + cancelRequested=True, + cancelTerminationActions=actions, + ) + response = jsonify(jobs.public_payload(updated)) + response.status_code = 202 + response.headers["Retry-After"] = "1" + return response + + +def _send_job_file(job_id, field, unavailable, mimetype, download_name=None): + job = jobs.load_job(job_id) + if job is None: + return jsonify({"error": f"Backtest job not found: {job_id}"}), 404 + path = job.get(field) + if not path or not os.path.isfile(path): + return jsonify({"error": unavailable, "jobId": job_id, "field": field}), 404 + return send_file( + path, + mimetype=mimetype, + as_attachment=False, + download_name=download_name or os.path.basename(path), + ) + + +def get_project_manifest(job_id): + return _send_job_file( + job_id, + "projectManifestPath", + "Project manifest not available", + "application/json", + ) + + +def get_project_bundle(job_id): + return _send_job_file( + job_id, + "projectBundlePath", + "Project bundle not available", + "application/zip", + ) + + +def get_submitted_ini(job_id): + return _send_job_file( + job_id, + "submittedIniPath", + "Submitted INI not available", + "text/plain", + ) + + +def get_execution_manifest(job_id): + return _send_job_file( + job_id, + "executionManifestPath", + "Execution manifest not available yet", + "application/json", + ) + + +def get_materialization_audit(job_id): + return _send_job_file( + job_id, + "materializationAuditPath", + "Materialization audit not available yet", + "application/json", + ) + + +def get_output_manifest(job_id): + return _send_job_file( + job_id, + "outputManifestPath", + "Output manifest not available yet", + "application/json", + ) + + +def get_output_artifacts(job_id): + return _send_job_file( + job_id, + "outputArtifactsPath", + "Output artifacts not available yet", + "application/zip", + ) + + +def get_log_delta_manifest(job_id): + return _send_job_file( + job_id, + "logDeltaManifestPath", + "Log delta manifest not available yet", + "application/json", + ) + + +def get_log_delta_artifacts(job_id): + return _send_job_file( + job_id, + "logDeltaArtifactsPath", + "Log delta artifacts not available yet", + "application/zip", + ) + + def get_report(job_id): job = jobs.load_job(job_id) if job is None: @@ -648,6 +1541,119 @@ def _tail_dir_log(log_dir, lines): return path, "\n".join(tail_lines[-lines:]) +def _decode_owned_log_bytes(path, raw): + with open(path, "rb") as handle: + prefix = handle.read(2) + if prefix == b"\xff\xfe": + payload = raw[2:] if raw.startswith(b"\xff\xfe") else raw + try: + return payload.decode("utf-16-le"), "utf-16-le", None, None + except UnicodeDecodeError as exc: + return None, None, base64.b64encode(raw).decode("ascii"), str(exc) + if prefix == b"\xfe\xff": + payload = raw[2:] if raw.startswith(b"\xfe\xff") else raw + try: + return payload.decode("utf-16-be"), "utf-16-be", None, None + except UnicodeDecodeError as exc: + return None, None, base64.b64encode(raw).decode("ascii"), str(exc) + try: + return raw.decode("utf-8"), "utf-8", None, None + except UnicodeDecodeError as exc: + return None, None, base64.b64encode(raw).decode("ascii"), str(exc) + + +def _owned_project_log_tail(job, n_lines): + records = [] + if job.get("logDeltaManifestPath") and os.path.isfile(job["logDeltaManifestPath"]): + with open(job["logDeltaManifestPath"], "r", encoding="utf-8") as handle: + stored = json.load(handle) + candidates = [ + { + "key": record["key"], + "classification": record["classification"], + "path": record.get("deltaPath"), + "rawSize": record.get("deltaSize"), + "rawSha256": record.get("deltaSha256"), + } + for record in stored.get("files", []) + if record.get("deltaPath") and os.path.isfile(record["deltaPath"]) + ] + source = "captured_delta" + elif job.get("logBaselinePath") and os.path.isfile(job["logBaselinePath"]): + with open(job["logBaselinePath"], "r", encoding="utf-8") as handle: + baseline_payload = json.load(handle) + baseline = baseline_payload.get("files") or {} + current = _log_inventory() + candidates = [] + for key, after in current.items(): + before = baseline.get(key) + offset = 0 + classification = "created" + if before is not None: + if after["size"] >= before["size"]: + prefix_sha256, prefix_size = _hash_prefix(after["path"], before["size"]) + if prefix_size == before["size"] and prefix_sha256 == before["sha256"]: + offset = before["size"] + classification = "appended" + else: + classification = "replaced" + else: + classification = "truncated_or_replaced" + if after["size"] == offset: + continue + candidates.append({ + "key": key, + "classification": classification, + "path": after["path"], + "offset": offset, + "rawSize": after["size"] - offset, + "rawSha256": None, + }) + source = "live_from_owned_baseline" + else: + return { + "source": "owned_baseline_unavailable", + "baselineAvailable": False, + "records": [], + "terminalLog": "", + "testerLog": "", + } + + grouped = {"terminal": [], "tester": []} + for candidate in candidates: + path = candidate["path"] + offset = candidate.get("offset", 0) + with open(path, "rb") as handle: + handle.seek(offset) + raw = handle.read() + raw_sha256 = hashlib.sha256(raw).hexdigest() + text, encoding, raw_base64, decode_error = _decode_owned_log_bytes(path, raw) + tail_text = "" + if text is not None: + lines = [line.strip() for line in text.splitlines() if line.strip()] + tail_text = "\n".join(lines[-n_lines:]) + scope = candidate["key"].split("/", 1)[0] + if tail_text and scope in grouped: + grouped[scope].append(tail_text) + records.append({ + "key": candidate["key"], + "classification": candidate["classification"], + "rawSize": len(raw), + "rawSha256": raw_sha256, + "encoding": encoding, + "text": tail_text if text is not None else None, + "rawBase64": raw_base64, + "decodeError": decode_error, + }) + return { + "source": source, + "baselineAvailable": True, + "records": records, + "terminalLog": "\n".join(grouped["terminal"]), + "testerLog": "\n".join(grouped["tester"]), + } + + def get_tail(job_id): """Live log tail — works for queued/running/completed jobs. @@ -674,6 +1680,24 @@ def get_tail(job_id): run_tail = [ln.strip() for ln in content.splitlines() if ln.strip()] run_log = "\n".join(run_tail[-50:]) + if job.get("projectMode"): + owned = _owned_project_log_tail(job, n_lines) + return jsonify({ + "jobId": job_id, + "externalRunId": job.get("externalRunId"), + "projectId": job.get("projectId"), + "status": job.get("status"), + "startedAt": job.get("startedAt"), + "finishedAt": job.get("finishedAt"), + "error": job.get("error"), + "runLog": run_log, + "terminalLog": owned["terminalLog"], + "testerLog": owned["testerLog"], + "logEvidenceSource": owned["source"], + "logBaselineAvailable": owned["baselineAvailable"], + "logEvidence": owned["records"], + }) + # MT5 terminal journal: /logs/YYYYMMDD.log terminal_log_dir = os.path.join(TERMINAL_DIR, "logs") terminal_log_file, terminal_log = _tail_dir_log(terminal_log_dir, n_lines) diff --git a/mt5api/backtest/jobs.py b/mt5api/backtest/jobs.py index e730e46..b37e57e 100644 --- a/mt5api/backtest/jobs.py +++ b/mt5api/backtest/jobs.py @@ -20,8 +20,9 @@ BACKTEST_JOBS: dict = {} POLL_AFTER_SECONDS = 60 -TERMINAL_STATUSES = frozenset({"completed", "failed"}) -ACTIVE_STATUSES = frozenset({"queued", "running"}) +PROJECT_POLL_AFTER_SECONDS = 2 +TERMINAL_STATUSES = frozenset({"completed", "failed", "canceled"}) +ACTIVE_STATUSES = frozenset({"queued", "running", "canceling"}) def now_iso() -> str: @@ -103,11 +104,32 @@ def public_payload(job: dict) -> dict: "reportUrl": f"/backtest/{job['jobId']}/report", "logUrl": f"/backtest/{job['jobId']}/log", "statusUrl": f"/backtest/{job['jobId']}", - "pollAfterSeconds": POLL_AFTER_SECONDS, + "pollAfterSeconds": ( + PROJECT_POLL_AFTER_SECONDS if job.get("projectMode") else POLL_AFTER_SECONDS + ), "optimizationType": job.get("optimizationType", 0), "optimizationResults": job.get("optimizationResults"), "optimizationCache": job.get("optimizationCache"), } + if job.get("projectMode"): + payload.update({ + "projectMode": True, + "projectId": job.get("projectId"), + "externalRunId": job.get("externalRunId"), + "planFingerprint": job.get("planFingerprint"), + "submittedIniUrl": f"/backtest/{job['jobId']}/submitted-ini", + "projectManifestUrl": f"/backtest/{job['jobId']}/project-manifest", + "projectBundleUrl": f"/backtest/{job['jobId']}/project-bundle", + "executionManifestUrl": f"/backtest/{job['jobId']}/execution-manifest", + "materializationAuditUrl": f"/backtest/{job['jobId']}/materialization", + "outputManifestUrl": f"/backtest/{job['jobId']}/output-manifest", + "outputArtifactsUrl": f"/backtest/{job['jobId']}/artifacts", + "logDeltaManifestUrl": f"/backtest/{job['jobId']}/log-deltas-manifest", + "logDeltaArtifactsUrl": f"/backtest/{job['jobId']}/log-deltas", + "cancelUrl": f"/backtest/{job['jobId']}", + "cancelRequested": bool(job.get("cancelRequested")), + "materializationStatus": job.get("materializationStatus"), + }) pos = queue_position(job["jobId"]) if pos is not None: payload["queuePosition"] = pos diff --git a/mt5api/backtest/materialization.py b/mt5api/backtest/materialization.py new file mode 100644 index 0000000..62c776a --- /dev/null +++ b/mt5api/backtest/materialization.py @@ -0,0 +1,797 @@ +"""Byte-exact project materialization for mt5-cli backtest jobs. + +This module deliberately does not infer missing project artifacts. A caller +must provide a complete manifest plus a ZIP containing every listed byte. The +module verifies the manifest, stages every payload, replaces only the declared +project module directories while the caller holds the terminal run lock, then +restores the prior directories byte for byte. +""" +from __future__ import annotations + +import hashlib +import json +import os +import re +import shutil +import stat +import zipfile +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import PurePosixPath + + +SCHEMA_VERSION = "MT5-CLI-PROJECT-MATERIALIZATION-001" +OUTPUT_SCHEMA_VERSION = "MT5-CLI-PROJECT-OUTPUTS-001" +AUDIT_SCHEMA_VERSION = "MT5-CLI-PROJECT-MATERIALIZATION-AUDIT-001" + +MODULE_ROOT_PARTS = { + "EXP": ("MQL5", "Experts"), + "IND": ("MQL5", "Indicators"), + "SCR": ("MQL5", "Scripts"), + "SRV": ("MQL5", "Services"), + "INC": ("MQL5", "Include"), + "LBR": ("MQL5", "Libraries"), + "PRS": ("MQL5", "Presets"), + "TPL": ("Profiles", "Templates"), +} +MODULES = frozenset((*MODULE_ROOT_PARTS.keys(), "FLS")) +_SHA256_RE = re.compile(r"^[0-9a-fA-F]{64}$") +_IDENTITY_RE = re.compile(r"^[A-Za-z0-9._:-]{1,200}$") + + +class MaterializationError(ValueError): + """A factual validation or filesystem failure with an explicit code.""" + + def __init__(self, code: str, message: str, *, details=None): + super().__init__(message) + self.code = code + self.details = details + + def as_dict(self) -> dict: + payload = {"code": self.code, "message": str(self)} + if self.details is not None: + payload["details"] = self.details + return payload + + +@dataclass(frozen=True) +class Artifact: + module: str + relative_path: str + bundle_entry: str + size: int + sha256: str + + @property + def parts(self): + return PurePosixPath(self.relative_path).parts + + @property + def module_directory(self) -> str: + return self.parts[0] + + @property + def path_inside_module(self) -> str: + return PurePosixPath(*self.parts[1:]).as_posix() + + +@dataclass(frozen=True) +class ProjectManifest: + raw: dict + project_id: str + run_id: str + plan_fingerprint: str + expert_relative_path: str + preset_relative_path: str | None + artifacts: tuple[Artifact, ...] + + @property + def module_directories(self) -> dict[str, str]: + result = {} + for artifact in self.artifacts: + previous = result.setdefault(artifact.module, artifact.module_directory) + if previous != artifact.module_directory: + raise MaterializationError( + "multiple_module_directories", + f"Module {artifact.module} contains more than one project directory", + details={ + "module": artifact.module, + "first": previous, + "second": artifact.module_directory, + }, + ) + return result + + +def _now_iso() -> str: + return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +def sha256_file(path: str) -> str: + digest = hashlib.sha256() + with open(path, "rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _write_json_atomic(path: str, value: dict) -> None: + os.makedirs(os.path.dirname(path), exist_ok=True) + temporary = f"{path}.tmp" + with open(temporary, "w", encoding="utf-8", newline="\n") as handle: + json.dump(value, handle, indent=2, sort_keys=True, ensure_ascii=False) + handle.write("\n") + os.replace(temporary, path) + + +def _require_identity(value, field: str) -> str: + if not isinstance(value, str) or not _IDENTITY_RE.fullmatch(value): + raise MaterializationError( + "invalid_identity", + f"{field} must match {_IDENTITY_RE.pattern}", + details={"field": field, "value": value}, + ) + return value + + +def _safe_relative_path(value, field: str) -> str: + if not isinstance(value, str) or not value: + raise MaterializationError( + "invalid_relative_path", + f"{field} must be a non-empty POSIX relative path", + details={"field": field, "value": value}, + ) + if "\\" in value: + raise MaterializationError( + "non_canonical_relative_path", + f"{field} must use forward slashes", + details={"field": field, "value": value}, + ) + path = PurePosixPath(value) + if path.is_absolute() or any(part in ("", ".", "..") for part in path.parts): + raise MaterializationError( + "unsafe_relative_path", + f"{field} is not a safe relative path", + details={"field": field, "value": value}, + ) + canonical = path.as_posix() + if canonical != value: + raise MaterializationError( + "non_canonical_relative_path", + f"{field} is not canonical", + details={"field": field, "value": value, "canonical": canonical}, + ) + return canonical + + +def _parse_artifact(raw, index: int) -> Artifact: + if not isinstance(raw, dict): + raise MaterializationError( + "invalid_artifact", + f"artifacts[{index}] must be an object", + details={"index": index}, + ) + module = raw.get("module") + if module not in MODULES: + raise MaterializationError( + "invalid_module", + f"artifacts[{index}].module is not supported", + details={"index": index, "module": module, "supported": sorted(MODULES)}, + ) + relative_path = _safe_relative_path( + raw.get("relativePath"), f"artifacts[{index}].relativePath" + ) + bundle_entry = _safe_relative_path( + raw.get("bundleEntry"), f"artifacts[{index}].bundleEntry" + ) + parts = PurePosixPath(relative_path).parts + if len(parts) < 2 or not parts[0].startswith(f"{module}-"): + raise MaterializationError( + "module_directory_mismatch", + f"artifacts[{index}].relativePath must begin with {module}-/", + details={"index": index, "module": module, "relativePath": relative_path}, + ) + size = raw.get("size") + if isinstance(size, bool) or not isinstance(size, int) or size < 0: + raise MaterializationError( + "invalid_artifact_size", + f"artifacts[{index}].size must be a non-negative integer", + details={"index": index, "size": size}, + ) + sha256 = raw.get("sha256") + if not isinstance(sha256, str) or not _SHA256_RE.fullmatch(sha256): + raise MaterializationError( + "invalid_artifact_sha256", + f"artifacts[{index}].sha256 must contain 64 hexadecimal characters", + details={"index": index, "sha256": sha256}, + ) + return Artifact(module, relative_path, bundle_entry, size, sha256.lower()) + + +def load_project_manifest(path: str) -> ProjectManifest: + try: + with open(path, "r", encoding="utf-8") as handle: + raw = json.load(handle) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: + raise MaterializationError( + "manifest_unreadable", + f"Project manifest cannot be read: {exc}", + details={"path": path, "errorType": type(exc).__name__}, + ) from exc + if not isinstance(raw, dict): + raise MaterializationError("manifest_not_object", "Project manifest must be a JSON object") + if raw.get("schemaVersion") != SCHEMA_VERSION: + raise MaterializationError( + "manifest_schema_mismatch", + f"schemaVersion must be {SCHEMA_VERSION}", + details={"actual": raw.get("schemaVersion"), "required": SCHEMA_VERSION}, + ) + project_id = _require_identity(raw.get("projectId"), "projectId") + run_id = _require_identity(raw.get("runId"), "runId") + plan_fingerprint = _require_identity(raw.get("planFingerprint"), "planFingerprint") + artifacts_raw = raw.get("artifacts") + if not isinstance(artifacts_raw, list) or not artifacts_raw: + raise MaterializationError( + "artifacts_absent", "Project manifest must contain at least one artifact" + ) + artifacts = tuple(_parse_artifact(item, index) for index, item in enumerate(artifacts_raw)) + + target_keys = set() + bundle_keys = set() + artifact_paths = {} + for artifact in artifacts: + target_key = (artifact.module, artifact.relative_path.casefold()) + bundle_key = artifact.bundle_entry.casefold() + if target_key in target_keys: + raise MaterializationError( + "duplicate_artifact_destination", + "Project manifest contains duplicate Windows destinations", + details={"module": artifact.module, "relativePath": artifact.relative_path}, + ) + if bundle_key in bundle_keys: + raise MaterializationError( + "duplicate_bundle_entry", + "Project manifest contains duplicate bundle entries", + details={"bundleEntry": artifact.bundle_entry}, + ) + target_keys.add(target_key) + bundle_keys.add(bundle_key) + artifact_paths[artifact.relative_path.casefold()] = artifact + + expert_relative_path = _safe_relative_path(raw.get("expertRelativePath"), "expertRelativePath") + expert_artifact = artifact_paths.get(expert_relative_path.casefold()) + if ( + expert_artifact is None + or expert_artifact.module != "EXP" + or not expert_relative_path.lower().endswith(".ex5") + ): + raise MaterializationError( + "expert_artifact_absent", + "expertRelativePath must identify a listed EXP .ex5 artifact", + details={"expertRelativePath": expert_relative_path}, + ) + + preset_value = raw.get("presetRelativePath") + preset_relative_path = None + if preset_value not in (None, ""): + preset_relative_path = _safe_relative_path(preset_value, "presetRelativePath") + preset_artifact = artifact_paths.get(preset_relative_path.casefold()) + if ( + preset_artifact is None + or preset_artifact.module != "PRS" + or not preset_relative_path.lower().endswith(".set") + ): + raise MaterializationError( + "preset_artifact_absent", + "presetRelativePath must identify a listed PRS .set artifact", + details={"presetRelativePath": preset_relative_path}, + ) + + manifest = ProjectManifest( + raw=raw, + project_id=project_id, + run_id=run_id, + plan_fingerprint=plan_fingerprint, + expert_relative_path=expert_relative_path, + preset_relative_path=preset_relative_path, + artifacts=artifacts, + ) + manifest.module_directories + return manifest + + +def _snapshot_tree(root: str) -> dict[str, dict]: + if not os.path.exists(root): + return {} + if not os.path.isdir(root): + raise MaterializationError( + "module_root_not_directory", + "A project module destination exists but is not a directory", + details={"path": root}, + ) + snapshot = {} + for current_root, directory_names, file_names in os.walk(root, followlinks=False): + for directory_name in directory_names: + directory_path = os.path.join(current_root, directory_name) + if _is_reparse(directory_path): + raise MaterializationError( + "reparse_point_in_module", + "A project module contains a symbolic link or junction", + details={"path": directory_path}, + ) + for file_name in file_names: + file_path = os.path.join(current_root, file_name) + if _is_reparse(file_path): + raise MaterializationError( + "reparse_point_in_module", + "A project module contains a symbolic link or junction", + details={"path": file_path}, + ) + relative = os.path.relpath(file_path, root).replace(os.sep, "/") + snapshot[relative] = { + "size": os.path.getsize(file_path), + "sha256": sha256_file(file_path), + } + return dict(sorted(snapshot.items(), key=lambda item: item[0].casefold())) + + +def _is_reparse(path: str) -> bool: + if os.path.islink(path): + return True + isjunction = getattr(os.path, "isjunction", None) + return bool(isjunction and isjunction(path)) + + +def _remove_path(path: str) -> None: + if not os.path.lexists(path): + return + if _is_reparse(path) or os.path.isfile(path): + os.unlink(path) + else: + shutil.rmtree(path) + + +def _assert_under_root(path: str, root: str) -> None: + absolute_path = os.path.abspath(path) + absolute_root = os.path.abspath(root) + try: + common = os.path.commonpath((absolute_path, absolute_root)) + except ValueError as exc: + raise MaterializationError( + "destination_outside_root", + "Project destination and module root are on different volumes", + details={"destination": absolute_path, "root": absolute_root}, + ) from exc + if os.path.normcase(common) != os.path.normcase(absolute_root): + raise MaterializationError( + "destination_outside_root", + "Project destination is outside its declared module root", + details={"destination": absolute_path, "root": absolute_root}, + ) + + +class ProjectMaterialization: + """Validated staged project with explicit apply/capture/restore lifecycle.""" + + def __init__( + self, + manifest_path: str, + bundle_path: str, + stage_dir: str, + terminal_dir: str, + common_files_dir: str | None, + ): + self.manifest_path = os.path.abspath(manifest_path) + self.bundle_path = os.path.abspath(bundle_path) + self.stage_dir = os.path.abspath(stage_dir) + self.terminal_dir = os.path.abspath(terminal_dir) + self.common_files_dir = os.path.abspath(common_files_dir) if common_files_dir else None + self.work_dir = os.path.join(self.stage_dir, "project-materialization") + self.payload_dir = os.path.join(self.work_dir, "payload") + self.module_stage_dir = os.path.join(self.work_dir, "modules") + self.backup_dir = os.path.join(self.work_dir, "backups") + self.audit_path = os.path.join(self.stage_dir, "materialization-audit.json") + self.output_manifest_path = os.path.join(self.stage_dir, "output-manifest.json") + self.output_archive_path = os.path.join(self.stage_dir, "output-artifacts.zip") + self.manifest = load_project_manifest(self.manifest_path) + self.staged_payloads: dict[str, str] = {} + self.module_targets: dict[str, str] = {} + self.module_stage_roots: dict[str, str] = {} + self.backup_snapshots: dict[str, dict] = {} + self.backup_present: dict[str, bool] = {} + self.applied_modules: list[str] = [] + self.audit = { + "schemaVersion": AUDIT_SCHEMA_VERSION, + "projectId": self.manifest.project_id, + "runId": self.manifest.run_id, + "planFingerprint": self.manifest.plan_fingerprint, + "createdAt": _now_iso(), + "manifestPath": self.manifest_path, + "manifestSha256": sha256_file(self.manifest_path), + "bundlePath": self.bundle_path, + "bundleSha256": sha256_file(self.bundle_path), + "terminalDir": self.terminal_dir, + "commonFilesDir": self.common_files_dir, + "status": "validating", + "artifacts": [], + "modules": {}, + "errors": [], + } + self._persist_audit() + try: + self._stage_and_verify_bundle() + self._build_module_trees() + self.audit["status"] = "validated" + self.audit["validatedAt"] = _now_iso() + self._persist_audit() + except Exception as exc: + self._record_error("validation_failed", exc) + raise + + @property + def expert_ini_value(self) -> str: + value = self.manifest.expert_relative_path[:-4] + return value.replace("/", "\\") + + @property + def preset_ini_value(self) -> str: + if not self.manifest.preset_relative_path: + return "" + return self.manifest.preset_relative_path.replace("/", "\\") + + def _persist_audit(self) -> None: + _write_json_atomic(self.audit_path, self.audit) + + def _record_error(self, phase: str, exc: Exception) -> None: + error = exc.as_dict() if isinstance(exc, MaterializationError) else { + "code": type(exc).__name__, + "message": str(exc), + } + self.audit["errors"].append({"phase": phase, "at": _now_iso(), **error}) + self.audit["status"] = phase + self._persist_audit() + + def _stage_and_verify_bundle(self) -> None: + os.makedirs(self.payload_dir, exist_ok=True) + try: + archive = zipfile.ZipFile(self.bundle_path, "r") + except (OSError, zipfile.BadZipFile) as exc: + raise MaterializationError( + "bundle_unreadable", + f"Project bundle is not a readable ZIP: {exc}", + details={"path": self.bundle_path, "errorType": type(exc).__name__}, + ) from exc + with archive: + infos = archive.infolist() + file_infos = {} + for info in infos: + if info.is_dir(): + directory_name = info.filename.rstrip("/") + _safe_relative_path(directory_name, "ZIP directory entry") + continue + canonical = _safe_relative_path(info.filename, "ZIP entry") + unix_type = (info.external_attr >> 16) & stat.S_IFMT(0o177777) + if unix_type == stat.S_IFLNK: + raise MaterializationError( + "bundle_symlink_rejected", + "Project bundle contains a symbolic link entry", + details={"bundleEntry": canonical}, + ) + key = canonical.casefold() + if key in file_infos: + raise MaterializationError( + "duplicate_zip_entry", + "Project bundle contains duplicate Windows entry names", + details={"bundleEntry": canonical}, + ) + file_infos[key] = (canonical, info) + + expected_keys = {artifact.bundle_entry.casefold() for artifact in self.manifest.artifacts} + actual_keys = set(file_infos) + if actual_keys != expected_keys: + raise MaterializationError( + "bundle_manifest_mismatch", + "Project bundle files do not exactly match manifest bundleEntry values", + details={ + "missing": sorted(expected_keys - actual_keys), + "unexpected": sorted(actual_keys - expected_keys), + }, + ) + + for index, artifact in enumerate(self.manifest.artifacts): + _, info = file_infos[artifact.bundle_entry.casefold()] + staged_path = os.path.join(self.payload_dir, f"{index:08d}.bin") + digest = hashlib.sha256() + size = 0 + with archive.open(info, "r") as source, open(staged_path, "wb") as destination: + while True: + chunk = source.read(1024 * 1024) + if not chunk: + break + destination.write(chunk) + digest.update(chunk) + size += len(chunk) + actual_sha256 = digest.hexdigest() + evidence = { + "module": artifact.module, + "relativePath": artifact.relative_path, + "bundleEntry": artifact.bundle_entry, + "declaredSize": artifact.size, + "actualSize": size, + "declaredSha256": artifact.sha256, + "actualSha256": actual_sha256, + "stagedPath": staged_path, + } + self.audit["artifacts"].append(evidence) + if size != artifact.size or actual_sha256 != artifact.sha256: + self._persist_audit() + raise MaterializationError( + "artifact_integrity_mismatch", + "A bundled project artifact does not match its declared size and SHA-256", + details=evidence, + ) + self.staged_payloads[artifact.relative_path.casefold()] = staged_path + self._persist_audit() + + def _module_base_root(self, module: str) -> str: + if module == "FLS": + if not self.common_files_dir: + raise MaterializationError( + "common_files_dir_absent", + "FLS artifacts require an explicit MetaTrader Common Files directory", + ) + return self.common_files_dir + return os.path.join(self.terminal_dir, *MODULE_ROOT_PARTS[module]) + + def _build_module_trees(self) -> None: + module_directories = self.manifest.module_directories + for module, module_directory in sorted(module_directories.items()): + base_root = self._module_base_root(module) + target = os.path.join(base_root, module_directory) + _assert_under_root(target, base_root) + if _is_reparse(target): + raise MaterializationError( + "module_target_reparse_point", + "Remote materialization requires a physical module directory, not a junction", + details={"module": module, "path": target}, + ) + staged_root = os.path.join(self.module_stage_dir, module, module_directory) + os.makedirs(staged_root, exist_ok=True) + self.module_targets[module] = target + self.module_stage_roots[module] = staged_root + self.audit["modules"][module] = { + "moduleDirectory": module_directory, + "baseRoot": base_root, + "targetPath": target, + "stagedRoot": staged_root, + "status": "staging", + } + + for artifact in self.manifest.artifacts: + destination = os.path.join( + self.module_stage_roots[artifact.module], + *PurePosixPath(artifact.path_inside_module).parts, + ) + os.makedirs(os.path.dirname(destination), exist_ok=True) + shutil.copyfile( + self.staged_payloads[artifact.relative_path.casefold()], destination + ) + + for module, staged_root in self.module_stage_roots.items(): + expected = { + artifact.path_inside_module: { + "size": artifact.size, + "sha256": artifact.sha256, + } + for artifact in self.manifest.artifacts + if artifact.module == module + } + actual = _snapshot_tree(staged_root) + if actual != expected: + raise MaterializationError( + "staged_module_integrity_mismatch", + "Staged module tree differs from the validated manifest", + details={"module": module, "expected": expected, "actual": actual}, + ) + self.audit["modules"][module]["status"] = "validated" + self.audit["modules"][module]["stagedSnapshot"] = actual + self._persist_audit() + + def apply(self) -> None: + if self.applied_modules: + raise MaterializationError( + "materialization_already_applied", + "Project materialization has already been applied", + details={"modules": list(self.applied_modules)}, + ) + self.audit["status"] = "backing_up" + self._persist_audit() + try: + for module, target in sorted(self.module_targets.items()): + base_root = os.path.dirname(target) + os.makedirs(base_root, exist_ok=True) + if _is_reparse(target): + raise MaterializationError( + "module_target_reparse_point", + "Remote materialization will not replace a junction or symbolic link", + details={"module": module, "path": target}, + ) + snapshot = _snapshot_tree(target) + self.backup_snapshots[module] = snapshot + present = os.path.exists(target) + self.backup_present[module] = present + backup = os.path.join(self.backup_dir, module) + if present: + os.makedirs(os.path.dirname(backup), exist_ok=True) + shutil.copytree(target, backup, copy_function=shutil.copyfile) + backup_snapshot = _snapshot_tree(backup) + if backup_snapshot != snapshot: + raise MaterializationError( + "backup_integrity_mismatch", + "Module backup differs from the pre-materialization tree", + details={ + "module": module, + "source": snapshot, + "backup": backup_snapshot, + }, + ) + self.audit["modules"][module]["backupPresent"] = present + self.audit["modules"][module]["backupPath"] = backup if present else None + self.audit["modules"][module]["backupSnapshot"] = snapshot + self.audit["modules"][module]["status"] = "backed_up" + self._persist_audit() + + self.audit["status"] = "applying" + self._persist_audit() + for module, target in sorted(self.module_targets.items()): + _remove_path(target) + shutil.copytree( + self.module_stage_roots[module], target, copy_function=shutil.copyfile + ) + applied_snapshot = _snapshot_tree(target) + staged_snapshot = self.audit["modules"][module]["stagedSnapshot"] + if applied_snapshot != staged_snapshot: + raise MaterializationError( + "applied_module_integrity_mismatch", + "Materialized module differs from its staged source", + details={ + "module": module, + "staged": staged_snapshot, + "applied": applied_snapshot, + }, + ) + self.applied_modules.append(module) + self.audit["modules"][module]["appliedSnapshot"] = applied_snapshot + self.audit["modules"][module]["status"] = "applied" + self._persist_audit() + self.audit["status"] = "applied" + self.audit["appliedAt"] = _now_iso() + self._persist_audit() + except Exception as exc: + self._record_error("apply_failed", exc) + try: + self.restore() + except Exception as restore_exc: + self._record_error("rollback_failed", restore_exc) + raise + + def capture_outputs(self) -> dict: + records = [] + output_files = [] + if "FLS" in self.module_targets: + target = self.module_targets["FLS"] + current = _snapshot_tree(target) + expected = self.audit["modules"]["FLS"]["stagedSnapshot"] + for relative_path in sorted( + set(current) | set(expected), key=lambda value: value.casefold() + ): + before = expected.get(relative_path) + after = current.get(relative_path) + if before is None: + classification = "created" + elif after is None: + classification = "missing_input" + elif before == after: + classification = "unchanged_input" + else: + classification = "modified" + record = { + "module": "FLS", + "relativePath": f"{self.manifest.module_directories['FLS']}/{relative_path}", + "classification": classification, + "input": before, + "output": after, + } + records.append(record) + if classification in ("created", "modified"): + output_files.append((relative_path, os.path.join(target, *PurePosixPath(relative_path).parts))) + + os.makedirs(os.path.dirname(self.output_archive_path), exist_ok=True) + with zipfile.ZipFile( + self.output_archive_path, "w", compression=zipfile.ZIP_DEFLATED, allowZip64=True + ) as archive: + for relative_path, source_path in output_files: + archive_name = ( + f"FLS/{self.manifest.module_directories['FLS']}/{relative_path}" + ) + archive.write(source_path, arcname=archive_name) + + output_manifest = { + "schemaVersion": OUTPUT_SCHEMA_VERSION, + "projectId": self.manifest.project_id, + "runId": self.manifest.run_id, + "planFingerprint": self.manifest.plan_fingerprint, + "capturedAt": _now_iso(), + "archivePath": self.output_archive_path, + "archiveSize": os.path.getsize(self.output_archive_path), + "archiveSha256": sha256_file(self.output_archive_path), + "files": records, + } + _write_json_atomic(self.output_manifest_path, output_manifest) + self.audit["outputManifestPath"] = self.output_manifest_path + self.audit["outputManifestSha256"] = sha256_file(self.output_manifest_path) + self.audit["outputArchivePath"] = self.output_archive_path + self.audit["outputArchiveSha256"] = output_manifest["archiveSha256"] + self.audit["outputsCapturedAt"] = output_manifest["capturedAt"] + self._persist_audit() + return output_manifest + + def restore(self) -> None: + failures = [] + modules_to_restore = [ + module + for module in sorted(self.module_targets, reverse=True) + if module in self.backup_present + ] + self.audit["status"] = "restoring" + self._persist_audit() + for module in modules_to_restore: + target = self.module_targets[module] + backup = os.path.join(self.backup_dir, module) + try: + _remove_path(target) + if self.backup_present[module]: + shutil.copytree(backup, target, copy_function=shutil.copyfile) + restored = _snapshot_tree(target) + expected = self.backup_snapshots[module] + if restored != expected: + raise MaterializationError( + "restore_integrity_mismatch", + "Restored module differs from its byte-exact backup", + details={ + "module": module, + "expected": expected, + "restored": restored, + }, + ) + elif os.path.exists(target): + raise MaterializationError( + "restore_absence_failed", + "A module that was absent before materialization still exists", + details={"module": module, "path": target}, + ) + self.audit["modules"][module]["status"] = "restored" + self.audit["modules"][module]["restoredSnapshot"] = ( + _snapshot_tree(target) if self.backup_present[module] else {} + ) + except Exception as exc: + error = exc.as_dict() if isinstance(exc, MaterializationError) else { + "code": type(exc).__name__, + "message": str(exc), + } + failures.append({"module": module, **error}) + self.audit["modules"][module]["status"] = "restore_failed" + self.audit["modules"][module]["restoreError"] = error + self._persist_audit() + self.applied_modules.clear() + if failures: + error = MaterializationError( + "restore_failed", + "One or more project modules could not be restored", + details=failures, + ) + self._record_error("restore_failed", error) + raise error + self.audit["status"] = "restored" + self.audit["restoredAt"] = _now_iso() + self._persist_audit() diff --git a/mt5api/config.py b/mt5api/config.py index 4370c90..d069c24 100644 --- a/mt5api/config.py +++ b/mt5api/config.py @@ -244,6 +244,17 @@ def load_terminal_config(): LOG_DIR = os.path.join(BASE_DIR, "logs") FULL_LOG = os.path.join(LOG_DIR, "full.log") BACKTEST_JOB_DIR = os.path.join(LOG_DIR, "backtest-jobs") +_COMMON_FILES_ENV = os.environ.get("MT5_COMMON_FILES_DIR") +_APPDATA_ENV = os.environ.get("APPDATA") +COMMON_FILES_DIR = ( + os.path.abspath(_COMMON_FILES_ENV) + if _COMMON_FILES_ENV + else ( + os.path.join(_APPDATA_ENV, "MetaQuotes", "Terminal", "Common", "Files") + if _APPDATA_ENV + else None + ) +) TIMEFRAME_MAP = { "M1": mt5.TIMEFRAME_M1, diff --git a/mt5api/server.py b/mt5api/server.py index 1aca624..e94130e 100644 --- a/mt5api/server.py +++ b/mt5api/server.py @@ -99,6 +99,16 @@ def _end_request(response): app.post("/backtest/build-set")(backtest_handler.build_set_route) app.post("/backtest")(backtest_handler.run_backtest) app.get("/backtest/")(backtest_handler.get_status) +app.delete("/backtest/")(backtest_handler.cancel_job) app.get("/backtest//report")(backtest_handler.get_report) app.get("/backtest//log")(backtest_handler.get_log) app.get("/backtest//tail")(backtest_handler.get_tail) +app.get("/backtest//submitted-ini")(backtest_handler.get_submitted_ini) +app.get("/backtest//project-manifest")(backtest_handler.get_project_manifest) +app.get("/backtest//project-bundle")(backtest_handler.get_project_bundle) +app.get("/backtest//execution-manifest")(backtest_handler.get_execution_manifest) +app.get("/backtest//materialization")(backtest_handler.get_materialization_audit) +app.get("/backtest//output-manifest")(backtest_handler.get_output_manifest) +app.get("/backtest//artifacts")(backtest_handler.get_output_artifacts) +app.get("/backtest//log-deltas-manifest")(backtest_handler.get_log_delta_manifest) +app.get("/backtest//log-deltas")(backtest_handler.get_log_delta_artifacts) diff --git a/tests/test_project_materialization.py b/tests/test_project_materialization.py new file mode 100644 index 0000000..101add4 --- /dev/null +++ b/tests/test_project_materialization.py @@ -0,0 +1,151 @@ +import hashlib +import json +from pathlib import Path +import zipfile + +import pytest + +from mt5api.backtest.materialization import ( + MaterializationError, + ProjectMaterialization, +) + + +def _sha256(payload: bytes) -> str: + return hashlib.sha256(payload).hexdigest() + + +def _write_project_contract(root: Path, *, omit_bundle_entry=None): + payloads = { + "payload/Expert.ex5": b"compiled-expert-real-bytes\x00\x01", + "payload/Expert.set": b"InpVolume=0.08||0.08||0.01||1.00||N\r\n", + "payload/input.csv": b"event,value\r\nbaseline,1\r\n", + } + artifacts = [ + { + "module": "EXP", + "relativePath": "EXP-Contract/Expert.ex5", + "bundleEntry": "payload/Expert.ex5", + }, + { + "module": "PRS", + "relativePath": "PRS-Contract/Expert.set", + "bundleEntry": "payload/Expert.set", + }, + { + "module": "FLS", + "relativePath": "FLS-Contract/input.csv", + "bundleEntry": "payload/input.csv", + }, + ] + for artifact in artifacts: + payload = payloads[artifact["bundleEntry"]] + artifact["size"] = len(payload) + artifact["sha256"] = _sha256(payload) + + manifest = { + "schemaVersion": "MT5-CLI-PROJECT-MATERIALIZATION-001", + "projectId": "project-real-contract", + "runId": "run-real-contract", + "planFingerprint": "plan-real-contract", + "expertRelativePath": "EXP-Contract/Expert.ex5", + "presetRelativePath": "PRS-Contract/Expert.set", + "artifacts": artifacts, + } + manifest_path = root / "project-manifest.json" + manifest_path.write_text( + json.dumps(manifest, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + bundle_path = root / "project-bundle.zip" + with zipfile.ZipFile(bundle_path, "w", compression=zipfile.ZIP_DEFLATED) as archive: + for entry, payload in payloads.items(): + if entry != omit_bundle_entry: + archive.writestr(entry, payload) + return manifest_path, bundle_path, payloads + + +def test_materialization_applies_captures_and_restores_real_bytes(tmp_path): + terminal_dir = tmp_path / "terminal" + common_files_dir = tmp_path / "common-files" + stage_dir = tmp_path / "job" + old_expert = terminal_dir / "MQL5" / "Experts" / "EXP-Contract" / "old.ex5" + old_common = common_files_dir / "FLS-Contract" / "preexisting.csv" + old_expert.parent.mkdir(parents=True) + old_common.parent.mkdir(parents=True) + old_expert.write_bytes(b"previous-expert-byte-exact") + old_common.write_bytes(b"previous,common,bytes\r\n") + manifest_path, bundle_path, payloads = _write_project_contract(tmp_path) + + materialization = ProjectMaterialization( + str(manifest_path), + str(bundle_path), + str(stage_dir), + str(terminal_dir), + str(common_files_dir), + ) + materialization.apply() + + applied_expert = terminal_dir / "MQL5" / "Experts" / "EXP-Contract" / "Expert.ex5" + applied_input = common_files_dir / "FLS-Contract" / "input.csv" + assert applied_expert.read_bytes() == payloads["payload/Expert.ex5"] + assert applied_input.read_bytes() == payloads["payload/input.csv"] + assert not old_expert.exists() + assert not old_common.exists() + + applied_input.write_bytes(b"event,value\r\nbaseline,2\r\n") + generated_output = common_files_dir / "FLS-Contract" / "trades.csv" + generated_output_payload = b"ticket,profit\r\n1001,12.34\r\n" + generated_output.write_bytes(generated_output_payload) + output_manifest = materialization.capture_outputs() + + classifications = { + item["relativePath"]: item["classification"] + for item in output_manifest["files"] + } + assert classifications == { + "FLS-Contract/input.csv": "modified", + "FLS-Contract/trades.csv": "created", + } + output_archive_path = Path(materialization.output_archive_path) + assert output_manifest["archiveSha256"] == _sha256(output_archive_path.read_bytes()) + with zipfile.ZipFile(output_archive_path, "r") as archive: + assert sorted(archive.namelist()) == [ + "FLS/FLS-Contract/input.csv", + "FLS/FLS-Contract/trades.csv", + ] + assert archive.read("FLS/FLS-Contract/trades.csv") == generated_output_payload + + materialization.restore() + assert old_expert.read_bytes() == b"previous-expert-byte-exact" + assert old_common.read_bytes() == b"previous,common,bytes\r\n" + assert not applied_expert.exists() + assert not applied_input.exists() + assert not generated_output.exists() + audit = json.loads(Path(materialization.audit_path).read_text(encoding="utf-8")) + assert audit["status"] == "restored" + assert audit["modules"]["EXP"]["restoredSnapshot"]["old.ex5"]["sha256"] == _sha256( + b"previous-expert-byte-exact" + ) + + +def test_materialization_rejects_bundle_missing_declared_real_file(tmp_path): + manifest_path, bundle_path, _ = _write_project_contract( + tmp_path, + omit_bundle_entry="payload/Expert.set", + ) + + with pytest.raises(MaterializationError) as failure: + ProjectMaterialization( + str(manifest_path), + str(bundle_path), + str(tmp_path / "job"), + str(tmp_path / "terminal"), + str(tmp_path / "common-files"), + ) + + assert failure.value.code == "bundle_manifest_mismatch" + assert failure.value.details == { + "missing": ["payload/expert.set"], + "unexpected": [], + }