Job Engineer¶

job_engineer — G6 Engineer job agent.

Cluster: Job Agents | Type: component | MCP Tools: 26

Overview¶

Domain-specialist job agent for software and systems engineers. Designs system architectures, reviews code quality, debugs issues from stack traces or logs, writes technical specifications, estimates effort, and analyses requirements — providing a structured engineering workflow within G6's safety-bounded, audit-trailed JobAgentBlock framework.

Paid-pilot engineering assistant, not licensed sign-off

job_engineer is suitable for paid pilots, local MCP workflows, code review assistance, specification drafting, and engineering-planning support where a qualified person remains accountable for the final decision. It is not autonomous engineering authority. Its calculations, standards checks, design reviews, and risk assessments are simplified reliability aids and must not be used as construction, fabrication, deployment, safety, regulatory, or professional-engineering sign-off without independent review by a licensed professional engineer or jurisdiction-equivalent qualified reviewer. Treat message and metadata.user_summary as user-facing summaries; inspect the structured result artifact and any metadata.integration_status degradation before relying on output in paid customer work.

Simulation access is exposed through parameters.simulation_request on design_system and benchmark_performance. It uses the in-memory analytical adapter only (structural_fea, thermal, modal, cfd), sets simulation_only=True, requires_review=True, and returns completion_state: qualified-draft unless invalid inputs require blocked-escalated. It is not a CAD/FEA/CFD solver substitute.

Public envelopes use the canonical completion states exactly: verified, qualified-draft, blocked-escalated. Advisory engineering outputs with missing optional integrations, simulation results, or human-review requirements surface qualified-draft plus a warning_card; refusal, invalid simulation input, or failure surfaces blocked-escalated.

When to use:

Reviewing a pull request or codebase for architectural issues, security anti-patterns, or maintainability
Generating a technical specification from high-level product requirements
Debugging a production issue by reasoning over logs, stack traces, and system state
Estimating effort and risk for a feature backlog or infrastructure migration

Example:

from mvp.job_engineer import JobEngineerBlock, JobEngineerInput

block = JobEngineerBlock()
result = block.infer(JobEngineerInput(
    task="Design a horizontally scalable microservice architecture for a real-time event-streaming platform handling 1M events/sec",
    context={"cloud": "AWS", "latency_p99_ms": 50, "team_size": 8},
))
# result.ok → True; result.value → JobEngineerOutput with result, artifacts

Works well with: job_framework, job_it, job_manager

Public API¶

`JobEngineerBlock(JobAgentBlock)`¶

G6 Engineer job agent -- Tier 1 block with MCP delegation.

Field	Type	Default
`name`	`str`	'job_engineer'
`sector`	`SectorClassification`	field(default_factory=lambda: _SECTOR)
`toolkit`	`ToolkitSpec \\| None`	field(default_factory=lambda: JOB_TOOLKITS.get('engineer'))
`mcp_module`	`str`	'mvp.job_engineer.engineer_mcp.server'
`agentic_planner`	`object \\| None`	None
`capabilities`	`ClassVar[set[type]]`	{Extensible, HumanLearnable, Collaborative, ProblemSolvable, KnowledgeGrounded, Memorable, AgentCommunicable, ExternallyAdaptable}

Methods:

`infer(data: Any) -> Result[JobEngineerOutput]`¶

`JobEngineerInput(JobInput)`¶

Input for the Engineer job agent.

`JobEngineerOutput(JobOutput)`¶

Output from the Engineer job agent.

`JobEngineerMCPBlock(AIBlock[MCPJobEngineerInput, MCPJobEngineerOutput, dict])`¶

26-op MCP block for the Engineer job agent.

Field	Type	Default
`name`	`str`	'job_engineer_mcp'
`state`	`dict`	field(default_factory=dict)
`db_path`	`str`	':memory:'
`resource_bounds`	`ResourceBounds`	field(default_factory=ResourceBounds)
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)

Methods:

`infer(data: MCPJobEngineerInput) -> Result[MCPJobEngineerOutput]`¶

`MCPJobEngineerInput(BaseModel)`¶

Input to JobEngineerMCPBlock -- 26-op dispatch.

Field	Type	Default
`op`	`Literal['design_system', 'review_code', 'debug_issue', 'write_spec', 'estimate_effort', 'analyze_requirements', 'evaluate_tradeoffs', 'assess_risk', 'benchmark_performance', 'audit_quality', 'create_proposal', 'review_deliverable', 'delegate_task', 'report_status', 'request_feedback', 'store_artifact', 'retrieve_artifact', 'list_artifacts', 'search_artifacts', 'archive', 'plan_sprint', 'track_progress', 'reflect_on_outcome', 'list_patterns', 'get_capabilities', 'info']`	required
`task`	`str`	''
`context`	`dict[str, Any]`	Field(default_factory=dict)
`parameters`	`dict[str, Any]`	Field(default_factory=dict)
`artifact_id`	`str`	''
`query`	`str`	''

`MCPJobEngineerOutput(BaseModel)`¶

Output from JobEngineerMCPBlock.

Field	Type	Default
`op`	`str`	required
`result`	`str`	''
`artifacts`	`list[dict[str, Any]]`	Field(default_factory=list)
`records`	`list[dict[str, Any]]`	Field(default_factory=list)
`message`	`str`	''
`count`	`int`	0
`found`	`bool`	False
`metadata`	`dict[str, Any]`	Field(default_factory=dict)
`degraded`	`bool`	False
`degradation_reason`	`str \\| None`	None
`completion_state`	`Literal['verified', 'qualified-draft', 'blocked-escalated']`	'qualified-draft'

`EngineerStore(JobStore)`¶

SQLite store for the Engineer job agent.

Constructor:

Parameter	Type	Default
`db_path`	`str`	':memory:'

Methods:

`save_design(project_name: str, system_type: str, data: dict, status: str = 'draft') -> str`¶

Persist a system design.

`get_design(design_id: str) -> dict | None`¶

`list_designs(status: str = '', limit: int = 50) -> list[dict]`¶

`update_design_status(design_id: str, status: str) -> bool`¶

`save_code_review(project_id: str, verdict: str, overall_score: float, data: dict) -> str`¶

Persist a code review result.

`get_code_reviews(project_id: str = '', limit: int = 20) -> list[dict]`¶

`save_risk_assessment(title: str, overall_level: str, data: dict) -> str`¶

Persist a risk assessment.

`get_risk_assessments(level: str = '', limit: int = 20) -> list[dict]`¶

`save_specification(spec_id: str, title: str, spec_type: str, data: dict) -> str`¶

Persist an engineering specification.

`get_specification(spec_id: str) -> dict | None`¶

`list_specifications(spec_type: str = '', limit: int = 50) -> list[dict]`¶

`update_spec_status(spec_id: str, status: str, revision: str = '') -> bool`¶

`save_benchmark(benchmark_type: str, data: dict) -> str`¶

Persist a benchmark result.

`get_benchmarks(benchmark_type: str = '', limit: int = 20) -> list[dict]`¶

`save_design_document(design_id: str, doc_type: str, title: str, content: dict, approved_by: str = '') -> str`¶

Persist a design document (drawing, calculation, etc.).

`list_design_documents(design_id: str = '', doc_type: str = '', limit: int = 50) -> list[dict]`¶

`count_all() -> dict[str, int]`¶

Override to include engineering-specific tables.

`engineering_summary() -> dict[str, Any]`¶

High-level summary of engineering activity.

Functions¶

`assemble_review_text(output: Any) -> str`¶

Collect the reviewable free text from a JobEngineerOutput (duck-typed).

`assess_engineer_output(output: Any, qa_block: Any | None = None, generate: Any | None = None) -> GroundedRunResult`¶

Run grounded four-valued QA over an engineer output.

MCP Tools¶

Operation	Source
`design_system`	engineer_mcp
`review_code`	engineer_mcp
`debug_issue`	engineer_mcp
`write_spec`	engineer_mcp
`estimate_effort`	engineer_mcp
`analyze_requirements`	engineer_mcp
`evaluate_tradeoffs`	engineer_mcp
`assess_risk`	engineer_mcp
`benchmark_performance`	engineer_mcp
`audit_quality`	engineer_mcp
`create_proposal`	engineer_mcp
`review_deliverable`	engineer_mcp
`delegate_task`	engineer_mcp
`report_status`	engineer_mcp
`request_feedback`	engineer_mcp
`store_artifact`	engineer_mcp
`retrieve_artifact`	engineer_mcp
`list_artifacts`	engineer_mcp
`search_artifacts`	engineer_mcp
`archive`	engineer_mcp
`plan_sprint`	engineer_mcp
`track_progress`	engineer_mcp
`reflect_on_outcome`	engineer_mcp
`list_patterns`	engineer_mcp
`get_capabilities`	engineer_mcp
`info`	engineer_mcp

Job Engineer¶

Overview¶

Public API¶

JobEngineerBlock(JobAgentBlock)¶

infer(data: Any) -> Result[JobEngineerOutput]¶

JobEngineerInput(JobInput)¶

JobEngineerOutput(JobOutput)¶

JobEngineerMCPBlock(AIBlock[MCPJobEngineerInput, MCPJobEngineerOutput, dict])¶

infer(data: MCPJobEngineerInput) -> Result[MCPJobEngineerOutput]¶

MCPJobEngineerInput(BaseModel)¶

MCPJobEngineerOutput(BaseModel)¶

EngineerStore(JobStore)¶

save_design(project_name: str, system_type: str, data: dict, status: str = 'draft') -> str¶

get_design(design_id: str) -> dict | None¶

list_designs(status: str = '', limit: int = 50) -> list[dict]¶

update_design_status(design_id: str, status: str) -> bool¶

save_code_review(project_id: str, verdict: str, overall_score: float, data: dict) -> str¶

get_code_reviews(project_id: str = '', limit: int = 20) -> list[dict]¶

save_risk_assessment(title: str, overall_level: str, data: dict) -> str¶

get_risk_assessments(level: str = '', limit: int = 20) -> list[dict]¶

save_specification(spec_id: str, title: str, spec_type: str, data: dict) -> str¶

get_specification(spec_id: str) -> dict | None¶

list_specifications(spec_type: str = '', limit: int = 50) -> list[dict]¶

update_spec_status(spec_id: str, status: str, revision: str = '') -> bool¶

save_benchmark(benchmark_type: str, data: dict) -> str¶

get_benchmarks(benchmark_type: str = '', limit: int = 20) -> list[dict]¶

save_design_document(design_id: str, doc_type: str, title: str, content: dict, approved_by: str = '') -> str¶

list_design_documents(design_id: str = '', doc_type: str = '', limit: int = 50) -> list[dict]¶

count_all() -> dict[str, int]¶

engineering_summary() -> dict[str, Any]¶

Functions¶

assemble_review_text(output: Any) -> str¶

assess_engineer_output(output: Any, qa_block: Any | None = None, generate: Any | None = None) -> GroundedRunResult¶

MCP Tools¶

`JobEngineerBlock(JobAgentBlock)`¶

`infer(data: Any) -> Result[JobEngineerOutput]`¶

`JobEngineerInput(JobInput)`¶

`JobEngineerOutput(JobOutput)`¶

`JobEngineerMCPBlock(AIBlock[MCPJobEngineerInput, MCPJobEngineerOutput, dict])`¶

`infer(data: MCPJobEngineerInput) -> Result[MCPJobEngineerOutput]`¶

`MCPJobEngineerInput(BaseModel)`¶

`MCPJobEngineerOutput(BaseModel)`¶

`EngineerStore(JobStore)`¶

`save_design(project_name: str, system_type: str, data: dict, status: str = 'draft') -> str`¶

`get_design(design_id: str) -> dict | None`¶

`list_designs(status: str = '', limit: int = 50) -> list[dict]`¶

`update_design_status(design_id: str, status: str) -> bool`¶

`save_code_review(project_id: str, verdict: str, overall_score: float, data: dict) -> str`¶

`get_code_reviews(project_id: str = '', limit: int = 20) -> list[dict]`¶

`save_risk_assessment(title: str, overall_level: str, data: dict) -> str`¶

`get_risk_assessments(level: str = '', limit: int = 20) -> list[dict]`¶

`save_specification(spec_id: str, title: str, spec_type: str, data: dict) -> str`¶

`get_specification(spec_id: str) -> dict | None`¶

`list_specifications(spec_type: str = '', limit: int = 50) -> list[dict]`¶

`update_spec_status(spec_id: str, status: str, revision: str = '') -> bool`¶

`save_benchmark(benchmark_type: str, data: dict) -> str`¶

`get_benchmarks(benchmark_type: str = '', limit: int = 20) -> list[dict]`¶

`save_design_document(design_id: str, doc_type: str, title: str, content: dict, approved_by: str = '') -> str`¶

`list_design_documents(design_id: str = '', doc_type: str = '', limit: int = 50) -> list[dict]`¶

`count_all() -> dict[str, int]`¶

`engineering_summary() -> dict[str, Any]`¶

`assemble_review_text(output: Any) -> str`¶

`assess_engineer_output(output: Any, qa_block: Any | None = None, generate: Any | None = None) -> GroundedRunResult`¶