347. 347 .arg(effective)
348. 348 .current_dir(&ctx.cwd)
349. 349 .stdin(Stdio::piped())
350. 350 .stdout(Stdio::piped())
351. 351 .stderr(Stdio::piped());
352. 352 crate::bash::scrub_environment(&mut cmd);
353. 353 cmd.env(WORKER_ENV, "1");
354. 354 crate::bash::isolate_process_group(&mut cmd);
355. 355
356. 356 let mut child = match cmd.spawn() {
357. 357 Ok(child) => child,
358. 358 Err(error) => return ToolOutput::error(format!("tool broker spawn failed: {error}")),
359. 359 };
360. 360 let request = WorkerRequest {
361. 361 version: PROTOCOL_VERSION,
362. 362 task: WorkerTask::Tool {
363. 363 tool: tool.to_string(),
364. 364 args: request_args,
365. 365 execution_id: ctx
366. 366 .sandbox_sink
367. 367 .as_ref()
368. 368 .map(|sink| sink.execution_id().to_string())
369. 369 .unwrap_or_else(|| "unidentified".into()),
370. 370 agent_id: ctx.agent_id.clone(),
371. 371 new_documents: new_documents.to_vec(),
372. 372 },
373. 373 };
374. 374 let payload = match serde_json::to_vec(&request) {
375. 375 Ok(payload) => payload,
376. 376 Err(error) => return ToolOutput::error(format!("tool broker encode failed: {error}")),
377. 377 };
378. 378 let Some(mut stdin) = child.stdin.take() else {
379. 379 crate::bash::kill_process_group(&child.id());
380. 380 return ToolOutput::error("tool broker has no stdin");
381. 381 };
382. 382 if let Err(error) = stdin.write_all(&payload).await {
383. 383 crate::bash::kill_process_group(&child.id());
384. 384 let _ = child.wait().await;
385. 385 return ToolOutput::error(format!("tool broker request failed: {error}"));
386. 386 }
387. 387 drop(stdin);
388. 388
389. 389 let Some(mut stdout) = child.stdout.take() else {
390. 390 return ToolOutput::error("tool broker has no stdout");
391. 391 };
392. 392 let Some(mut stderr) = child.stderr.take() else {
393. 393 return ToolOutput::error("tool broker has no stderr");
394. 394 };
395. 395 let sink = ctx.sandbox_sink.clone();
396. 396 let partial = Arc::new(std::sync::Mutex::new(String::new()));
397. 397 let partial_reader = partial.clone();
398. 398 let stderr_reader = tokio::spawn(async move {
399. 399 let mut bytes = Vec::new();
400. 400 let mut line = Vec::new();
401. 401 while let Ok(n) = stderr.read_buf(&mut line).await {
402. 402 if n == 0 {
403. 403 break;
404. 404 }
405. 405 while let Some(pos) = line.iter().position(|b| *b == b'\n') {
406. 406 let frame: Vec<u8> = line.drain(..=pos).collect();
407. 407 if let Some(payload) = frame.strip_prefix(b"VAK_EVENT:")
408. 408 && let Ok(event) = serde_json::from_slice::<SandboxEvent>(payload)
409. 409 {
410. 410 if let Ok(mut output) = partial_reader.lock() {
411. 411 match &event {
412. 412 SandboxEvent::Stdout { chunk, .. } => {
413. 413 output.push_str(chunk);
414. 414 }
415. 415 SandboxEvent::Stderr { chunk, .. } => {
416. 416 output.push_str("[stderr]\n");
417. 417 output.push_str(chunk);
418. 418 }
419. 419 _ => {}
420. 420 }
421. 421 }
422. 422 if let Some(ref sink) = sink {
423. 423 sink.emit(event);
424. 424 }
425. 425 } else {
426. 426 bytes.extend(frame);
427. 427 }
428. 428 }
429. 429 }
430. 430 bytes.extend(line);
431. 431 bytes
432. 432 });
433. 433 let stdout_reader = tokio::spawn(async move {
434. 434 let mut bytes = Vec::new();
435. 435 let _ = stdout.read_to_end(&mut bytes).await;
436. 436 bytes
437. 437 });
438. 438 let pid = child.id();
439. 439 let (status, cancelled) = tokio::select! {
440. 440 _ = ctx.cancel.cancelled() => {
441. 441 crate::bash::kill_process_group(&pid);
442. 442 (child.wait().await, true)
443. 443 }
444. 444 output = child.wait() => (output, false),
445. 445 };
446. 446 let status = match status {
447. 447 Ok(status) => status,
448. 448 Err(error) => return ToolOutput::error(format!("tool broker wait failed: {error}")),
449. 449 };
450. 450 let stdout = stdout_reader.await.unwrap_or_default();
451. 451 let stderr = stderr_reader.await.unwrap_or_default();
452. 452 if cancelled {
453. 453 let mut content = String::from("tool broker cancelled");
454. 454 if let Ok(output) = partial.lock()
455. 455 && !output.is_empty()
456. 456 {
457. 457 content.push_str("\n\n[partial output]\n");
458. 458 content.push_str(&output);
459. 459 }
460. 460 let diagnostics = String::from_utf8_lossy(&stderr).trim().to_string();
461. 461 if !diagnostics.is_empty() {
462. 462 content.push_str("\n\n[diagnostics]\n");
463. 463 content.push_str(&diagnostics);
464. 464 }
465. 465 let _ = stdout;
466. 466 return ToolOutput::error(content);
467. 467 }
468. 468 if stdout.len() as u64 > MAX_PROTOCOL_BYTES || stderr.len() as u64 > MAX_PROTOCOL_BYTES {
469. 469 return ToolOutput::error("tool broker output exceeded protocol limit");
470. 470 }
471. 471 if !status.success() {
472. 472 let detail = String::from_utf8_lossy(&stderr);
473. 473 return ToolOutput::error(format!(
474. 474 "tool broker exited with {}: {}",
475. 475 status.code().unwrap_or(-1),
476. 476 detail.trim()
477. 477 ));
478. 478 }
479. 479 let response: WorkerResponse = match serde_json::from_slice(&stdout) {
480. 480 Ok(response) => response,
481. 481 Err(error) => {
482. 482 let detail = String::from_utf8_lossy(&stderr);
483. 483 return ToolOutput::error(format!(
484. 484 "tool broker returned invalid protocol: {error}; stderr: {}",
485. 485 detail.trim()
486. 486 ));
487. 487 }
488. 488 };
489. 489 if response.version != PROTOCOL_VERSION {
490. 490 return ToolOutput::error(format!(
491. 491 "tool broker protocol mismatch: expected {PROTOCOL_VERSION}, got {}",
492. 492 response.version
493. 493 ));
494. 494 }
495. 495 if response.is_error {
496. 496 ToolOutput::error(response.content)
497. 497 } else {
498. 498 ToolOutput::ok(response.content)
499. 499 }
500. 500}
501. 501
502. 502pub async fn worker_main() -> i32 {
503. 503 let mut bytes = Vec::new();
504. 504 if tokio::io::stdin()
505. 505 .take(MAX_PROTOCOL_BYTES + 1)
506. 506 .read_to_end(&mut bytes)
507. 507 .await
508. 508 .is_err()
509. 509 || bytes.len() as u64 > MAX_PROTOCOL_BYTES
510. 510 {
511. 511 return 125;
512. 512 }
513. 513 let request: WorkerRequest = match serde_json::from_slice::<WorkerRequest>(&bytes) {
514. 514 Ok(request) if request.version == PROTOCOL_VERSION => request,
515. 515 _ => return 125,
516. 516 };
517. 517 let (tool_name, args, execution_id, agent_id, new_documents) = match request.task {
518. 518 WorkerTask::Tool {
519. 519 tool,
520. 520 args,
521. 521 execution_id,
522. 522 agent_id,
523. 523 new_documents,
524. 524 } => (tool, args, execution_id, agent_id, new_documents),
525. 525 WorkerTask::OfficeReview {
526. 526 before,
527. 527 after,
528. 528 lineage,
529. 529 } => {
530. 530 let (content, is_error) =
531. 531 match office_review_in_worker(before.as_deref(), &after, lineage.as_ref()) {
532. 532 Ok(content) => (content, false),
533. 533 Err(error) => (error, true),
534. 534 };
535. 535 return write_response(WorkerResponse {
536. 536 version: PROTOCOL_VERSION,
537. 537 content,
538. 538 is_error,
539. 539 events: Vec::new(),
540. 540 })
541. 541 .await;
542. 542 }
543. 543 WorkerTask::OfficeProject { path, view } => {
544. 544 let (content, is_error) = match office_project_in_worker(&path, view) {
545. 545 Ok(content) => (content, false),
546. 546 Err(error) => (error, true),
547. 547 };
548. 548 return write_response(WorkerResponse {
549. 549 version: PROTOCOL_VERSION,
550. 550 content,
551. 551 is_error,
552. 552 events: Vec::new(),
553. 553 })
554. 554 .await;
555. 555 }
556. 556 WorkerTask::OfficeApply { lineage, out } => {
557. 557 let (content, is_error) = match office_apply_in_worker(&lineage, &out) {
558. 558 Ok(content) => (content, false),
559. 559 Err(error) => (error, true),
560. 560 };
561. 561 return write_response(WorkerResponse {
562. 562 version: PROTOCOL_VERSION,
563. 563 content,
564. 564 is_error,
565. 565 events: Vec::new(),
566. 566 })
567. 567 .await;
568. 568 }
569. 569 WorkerTask::OfficeNarrow {
570. 570 lineage,
571. 571 draft,
572. 572 keep,
573. 573 out,
574. 574 } => {
575. 575 let (content, is_error) = match office_narrow_in_worker(&lineage, &draft, &keep, &out) {
576. 576 Ok(content) => (content, false),
577. 577 Err(error) => (error, true),
578. 578 };
579. 579 return write_response(WorkerResponse {
580. 580 version: PROTOCOL_VERSION,
581. 581 content,
582. 582 is_error,
583. 583 events: Vec::new(),
584. 584 })
585. 585 .await;
586. 586 }
587. 587 WorkerTask::VerifyTargets { root, checks } => {
588. 588 let results = vak_sandbox::default_target_verifiers().verify(&root, &checks);
589. 589 let content = serde_json::to_string(&results).unwrap_or_default();
590. 590 return write_response(WorkerResponse {
591. 591 version: PROTOCOL_VERSION,
592. 592 content,
593. 593 is_error: false,
594. 594 events: Vec::new(),
595. 595 })
596. 596 .await;
597. 597 }
598. 598 };
599. 599 let tool = crate::default_tools()
600. 600 .into_iter()
601. 601 .find(|candidate| candidate.name() == tool_name);
602. 602 let (output, events) = match tool {
603. 603 Some(tool) => {
604. 604 let cwd = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
605. 605 let (sink, mut rx) = crate::sandbox_events::SandboxEventSink::new_with_id(execution_id);
606. 606 let mut ctx = ToolContext::new(cwd)
607. 607 .with_sandbox_sink(sink)
608. 608 .with_new_documents(new_documents);
609. 609 if let Some(agent_id) = agent_id {
610. 610 ctx = ctx.with_agent_id(agent_id);
611. 611 }
612. 612 let event_forwarder = tokio::spawn(async move {
613. 613 let mut stderr = tokio::io::stderr();
614. 614 while let Some(event) = rx.recv().await {
615. 615 let Ok(mut frame) = serde_json::to_vec(&event) else {
616. 616 continue;
617. 617 };
618. 618 let mut prefixed = b"VAK_EVENT:".to_vec();
619. 619 prefixed.append(&mut frame);
620. 620 prefixed.push(b'\n');
621. 621 if stderr.write_all(&prefixed).await.is_err() {
622. 622 break;
623. 623 }
624. 624 let _ = stderr.flush().await;
625. 625 }
626. 626 });
627. 627 let output = tool.execute(&args, &ctx).await;
628. 628 drop(ctx.sandbox_sink);
629. 629 let _ = event_forwarder.await;
630. 630 (output, Vec::new())
631. 631 }
632. 632 None => (
633. 633 ToolOutput::error(format!("worker does not expose tool '{tool_name}'")),
634. 634 Vec::new(),
635. 635 ),
636. 636 };
637. 637 write_response(WorkerResponse {
638. 638 version: PROTOCOL_VERSION,
639. 639 content: output.content,
640. 640 is_error: output.is_error,
641. 641 events,
642. 642 })
643. 643 .await
644. 644}
645. 645
646. 646async fn write_response(response: WorkerResponse) -> i32 {
647. 647 let payload = match serde_json::to_vec(&response) {
648. 648 Ok(payload) if payload.len() as u64 <= MAX_PROTOCOL_BYTES => payload,
649. 649 Ok(payload) => {
650. 650 let refusal = WorkerResponse {
651. 651 version: PROTOCOL_VERSION,
652. 652 content: format!(
653. 653 "the result is {} bytes, over the {} byte worker protocol limit; request a smaller range",
654. 654 payload.len(),
655. 655 MAX_PROTOCOL_BYTES
656. 656 ),
657. 657 is_error: true,
658. 658 events: Vec::new(),
659. 659 };
660. 660 match serde_json::to_vec(&refusal) {
661. 661 Ok(payload) => payload,
662. 662 Err(_) => return 125,
663. 663 }
664. 664 }
665. 665 Err(_) => return 125,
666. 666 };
667. 667 let mut stdout = tokio::io::stdout();
668. 668 if stdout.write_all(&payload).await.is_err() || stdout.flush().await.is_err() {
669. 669 return 125;
670. 670 }
671. 671 0
672. 672}
673. 673
674. 674/// Runs the registered target verifiers over `checks` under `root` in a
675. 675/// worker process (docs/design/72-openxml-documents.md, F4). The worker runs
676. 676/// under a read-only, network-denied sandbox rooted at `root` whatever the
677. 677/// session's permission mode, and within [`VERIFY_DEADLINE`]. Anything that
678. 678/// stops the worker from answering fails every planned check; a check never
679. 679/// passes because verification could not run.
680. 680pub async fn verify_targets(
681. 681 worker_exe: &Path,
682. 682 root: &Path,
683. 683 checks: &[vak_sandbox::TargetCheckPlan],
684. 684) -> Vec<vak_sandbox::TargetCheckResult> {
685. 685 if checks.is_empty() {
686. 686 return Vec::new();
687. 687 }
688. 688 let failed = |reason: String| {
689. 689 checks
690. 690 .iter()
691. 691 .map(|check| vak_sandbox::TargetCheckResult {
692. 692 verifier: check.verifier.clone(),
693. 693 path: check.path.clone(),
694. 694 status: "failed".into(),
695. 695 evidence: format!("verification did not run: {reason}"),
696. 696 })
697. 697 .collect::<Vec<_>>()
698. 698 };
699. 699 let task = WorkerTask::VerifyTargets {
700. 700 root: root.to_path_buf(),
701. 701 checks: checks.to_vec(),
702. 702 };
703. 703 let content = match run_task(worker_exe, root, &[root], false, task).await {
704. 704 Ok(content) => content,
705. 705 Err(reason) => return failed(reason),
706. 706 };
707. 707 match serde_json::from_str::<Vec<vak_sandbox::TargetCheckResult>>(&content) {
708. 708 Ok(results) if results.len() == checks.len() => results,
709. 709 Ok(_) => failed("worker answered a different number of checks".into()),
710. 710 Err(error) => failed(format!("worker returned invalid results: {error}")),
711. 711 }
712. 712}
713. 713
714. 714/// What an Office draft (`after`) changes compared with the file it would
715. 715/// replace (`before`, absent for a new file), computed in a worker under the
716. 716/// same read-only sandbox and deadline as verification, because every file
717. 717/// involved is hostile input (invariant 39). With the draft's `lineage`, the
718. 718/// answer also carries the choices a person can keep or leave out
719. 719/// (`vak_ooxml::review`), or the reason the draft can only be taken whole.
720. 720pub async fn office_review(
721. 721 worker_exe: &Path,
722. 722 before: Option<&Path>,
723. 723 after: &Path,
724. 724 lineage: Option<&OfficeLineage>,
725. 725) -> Result<Value, String> {
726. 726 let Some(after_dir) = after.parent() else {
727. 727 return Err("the draft has no directory".into());
728. 728 };
729. 729 let mut roots = vec![after_dir];
730. 730 roots.extend(before.and_then(Path::parent));
731. 731 roots.extend(lineage.and_then(|lineage| lineage.origin.directory()));
732. 732 let task = WorkerTask::OfficeReview {
733. 733 before: before.map(Path::to_path_buf),
734. 734 after: after.to_path_buf(),
735. 735 lineage: lineage.cloned(),
736. 736 };
737. 737 let content = run_task(worker_exe, after_dir, &roots, false, task).await?;
738. 738 serde_json::from_str(&content)
739. 739 .map_err(|error| format!("worker returned an invalid review: {error}"))
740. 740}
741. 741
742. 742/// Replays the choices in `keep` from `lineage` and writes the narrower
743. 743/// version of `draft` to `out`, refusing a draft the lineage does not
744. 744/// reproduce. The worker can write only `out`'s directory, which must
745. 745/// exist, and read only the lineage's source and the draft.
746. 746pub async fn office_narrow(
747. 747 worker_exe: &Path,
748. 748 lineage: &OfficeLineage,
749. 749 draft: &Path,
750. 750 keep: &[String],
751. 751 out: &Path,
752. 752) -> Result<Value, String> {
753. 753 let Some(out_dir) = out.parent() else {
754. 754 return Err("the output has no directory".into());
755. 755 };
756. 756 let mut roots = vec![out_dir];
757. 757 roots.extend(lineage.origin.directory());
758. 758 roots.extend(draft.parent());
759. 759 let task = WorkerTask::OfficeNarrow {
760. 760 lineage: lineage.clone(),
761. 761 draft: draft.to_path_buf(),
762. 762 keep: keep.to_vec(),
763. 763 out: out.to_path_buf(),
764. 764 };
765. 765 let content = run_task(worker_exe, out_dir, &roots, true, task).await?;
766. 766 serde_json::from_str(&content)
767. 767 .map_err(|error| format!("worker returned an invalid answer: {error}"))
768. 768}
769. 769
770. 770/// Applies `lineage`'s ops to its source and writes the result to `out`, a
771. 771/// new file (`vak office apply`, docs/design/72 P5). The same checked apply
772. 772/// as the `office_apply` tool; the worker can write only `out`'s directory
773. 773/// and read only the source's. The answer lists each op's result and
774. 774/// postcondition, the engine's notices, the semantic change list against
775. 775/// the source, and what the edit does to signatures and labels.
776. 776pub async fn office_apply_to(
777. 777 worker_exe: &Path,
778. 778 lineage: &OfficeLineage,
779. 779 out: &Path,
780. 780) -> Result<Value, String> {
781. 781 let Some(out_dir) = out.parent() else {
782. 782 return Err("the output has no directory".into());
783. 783 };
784. 784 let mut roots = vec![out_dir];
785. 785 roots.extend(lineage.origin.directory());
786. 786 let task = WorkerTask::OfficeApply {
787. 787 lineage: lineage.clone(),
788. 788 out: out.to_path_buf(),
789. 789 };
790. 790 let content = run_task(worker_exe, out_dir, &roots, true, task).await?;
791. 791 serde_json::from_str(&content)
792. 792 .map_err(|error| format!("worker returned an invalid answer: {error}"))
793. 793}
794. 794
795. 795fn office_apply_in_worker(lineage: &OfficeLineage, out: &Path) -> Result<String, String> {
796. 796 if std::fs::symlink_metadata(out).is_ok() {
797. 797 return Err(format!(
798. 798 "{} already exists; name a new file for the result",
799. 799 out.display()
800. 800 ));
801. 801 }
802. 802 let target = out
803. 803 .extension()
804. 804 .and_then(|extension| extension.to_str())
805. 805 .and_then(vak_ooxml::Format::from_extension)
806. 806 .ok_or_else(|| {
807. 807 format!(
808. 808 "{} is not named as a Word, Excel or PowerPoint file",
809. 809 out.display()
810. 810 )
811. 811 })?;
812. 812 let context = lineage_context(lineage);
813. 813 let (source, applied) =
814. 814 crate::office_apply::apply_checked(&lineage.origin, &lineage.ops, &context, target)?;
815. 815 let before = match &lineage.origin {
816. 816 OfficeOrigin::File { path, .. } => Some(
817. 817 vak_ooxml::read::read(std::io::Cursor::new(source), vak_ooxml::Limits::default())
818. 818 .map_err(|error| format!("{} could not be read: {error}", path.display()))?,
819. 819 ),
820. 820 OfficeOrigin::Blank => None,
821. 821 };
822. 822 crate::office_apply::write_atomically(out, &applied.bytes)?;
823. 823 serde_json::to_string(&serde_json::json!({
824. 824 "path": out,
825. 825 "sha256": crate::office_apply::sha256_hex(&applied.bytes),
826. 826 "results": applied.results,
827. 827 "notices": applied.notices,
828. 828 "changes": vak_ooxml::diff::diff(before.as_ref(), &applied.document),
829. 829 "impact": vak_ooxml::diff::impact(before.as_ref(), &applied.document),
830. 830 }))
831. 831 .map_err(|error| error.to_string())
832. 832}
833. 833
834. 834/// What a view draws of the Office file at `path`: a page of its content
835. 835/// or its structure, with the file's digest so an anchor chosen in the
836. 836/// view is bound to these exact bytes. Parsed in a worker under the
837. 837/// read-only sandbox, like every Office read (invariant 39).
838. 838pub async fn office_project(
839. 839 worker_exe: &Path,
840. 840 path: &Path,
841. 841 view: OfficeView,
842. 842) -> Result<Value, String> {
843. 843 let Some(dir) = path.parent() else {
844. 844 return Err("the file has no directory".into());
845. 845 };
846. 846 let task = WorkerTask::OfficeProject {
847. 847 path: path.to_path_buf(),
848. 848 view,
849. 849 };
850. 850 let content = run_task(worker_exe, dir, &[dir], false, task).await?;
851. 851 serde_json::from_str(&content)
852. 852 .map_err(|error| format!("worker returned an invalid projection: {error}"))
853. 853}
854. 854
855. 855fn office_project_in_worker(path: &Path, view: OfficeView) -> Result<String, String> {
856. 856 let limits = vak_ooxml::Limits::default();
857. 857 let bytes = crate::office_apply::read_bounded(path, &limits)?;
858. 858 let sha256 = crate::office_apply::sha256_hex(&bytes);
859. 859 let mut body = match view {
860. 860 OfficeView::Content { .. } | OfficeView::At { .. } | OfficeView::Facts => {
861. 861 let document = vak_ooxml::read::read(std::io::Cursor::new(bytes), limits)
862. 862 .map_err(|error| format!("{} could not be read: {error}", path.display()))?;
863. 863 let focus = match &view {
864. 864 OfficeView::At { anchor } => vak_ooxml::projection::locate(&document, anchor),
865. 865 _ => None,
866. 866 };
867. 867 let from = match &view {
868. 868 OfficeView::Content { from } => *from,
869. 869 OfficeView::At { .. } => focus
870. 870 .map(|index| {
871. 871 vak_ooxml::projection::page_start(
872. 872 &document,
873. 873 index,
874. 874 vak_ooxml::projection::PAGE_BYTES,
875. 875 )
876. 876 })
877. 877 .unwrap_or(0),
878. 878 _ => document.units.len(),
879. 879 };
880. 880 let mut page = serde_json::to_value(vak_ooxml::projection::project(
881. 881 &document,
882. 882 from,
883. 883 vak_ooxml::projection::PAGE_BYTES,
884. 884 ));
885. 885 if let (Ok(page), Some(index)) = (&mut page, focus) {
886. 886 page["focus"] = Value::String(document.units[index].anchor.clone());
887. 887 }
888. 888 page
889. 889 }
890. 890 OfficeView::Structure => {
891. 891 let structure =
892. 892 vak_ooxml::projection::structure(std::io::Cursor::new(bytes), limits)
893. 893 .map_err(|error| format!("{} could not be read: {error}", path.display()))?;
894. 894 serde_json::to_value(structure)
895. 895 }
896. 896 }
897. 897 .map_err(|error| error.to_string())?;
898. 898 body["sha256"] = Value::String(sha256);
899. 899 serde_json::to_string(&body).map_err(|error| error.to_string())
900. 900}
901. 901
902. 902fn read_document(path: &Path) -> Result<vak_ooxml::read::Document, String> {
903. 903 let file = std::fs::File::open(path)
904. 904 .map_err(|error| format!("cannot open {}: {error}", path.display()))?;
905. 905 vak_ooxml::read::read(file, vak_ooxml::Limits::default())
906. 906 .map_err(|error| format!("{} could not be read: {error}", path.display()))
907. 907}
908. 908
909. 909/// The bytes the lineage starts from: its file, refused unless it still has
910. 910/// the digest the draft was made against, or the built-in blank for
911. 911/// `target`, the format the draft is named as.
912. 912fn lineage_source(
913. 913 lineage: &OfficeLineage,
914. 914 target: Option<vak_ooxml::Format>,
915. 915) -> Result<Vec<u8>, String> {
916. 916 let (path, base_digest) = match &lineage.origin {
917. 917 OfficeOrigin::File { path, base_digest } => (path, base_digest),
918. 918 OfficeOrigin::Blank => {
919. 919 let Some(target) = target else {
920. 920 return Err("the draft is not named as a Word, Excel or PowerPoint file".into());
921. 921 };
922. 922 return vak_ooxml::blank::blank(target);
923. 923 }
924. 924 };
925. 925 let limits = vak_ooxml::Limits::default();
926. 926 let bytes = crate::office_apply::read_bounded(path, &limits)?;
927. 927 let digest = crate::office_apply::sha256_hex(&bytes);
928. 928 let expected = base_digest
929. 929 .trim()
930. 930 .trim_end_matches('…')
931. 931 .to_ascii_lowercase();
932. 932 if expected.len() < 16 || !digest.starts_with(&expected) {
933. 933 return Err(format!(
934. 934 "{} changed after the draft was made from it",
935. 935 path.display()
936. 936 ));
937. 937 }
938. 938 Ok(bytes)
939. 939}
940. 940
941. 941fn lineage_context(lineage: &OfficeLineage) -> vak_ooxml::edit::EditContext {
942. 942 vak_ooxml::edit::EditContext {
943. 943 author: lineage.author.clone(),
944. 944 date: chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string(),
945. 945 tracked: !lineage.new_file && lineage.origin != OfficeOrigin::Blank,
946. 946 }
947. 947}
948. 948
949. 949fn target_of(path: &Path) -> Option<vak_ooxml::Format> {
950. 950 path.extension()
951. 951 .and_then(|extension| extension.to_str())
952. 952 .and_then(vak_ooxml::Format::from_extension)
953. 953}
954. 954
955. 955fn office_review_in_worker(
956. 956 before: Option<&Path>,
957. 957 after: &Path,
958. 958 lineage: Option<&OfficeLineage>,
959. 959) -> Result<String, String> {
960. 960 let current = before.map(read_document).transpose()?;
961. 961 let draft = read_document(after)?;
962. 962 let diff = vak_ooxml::diff::diff(current.as_ref(), &draft);
963. 963 let mut body = serde_json::json!({
964. 964 "summary": diff.summary,
965. 965 "changes": diff.changes,
966. 966 "flags": draft.inspection.flags(),
967. 967 "impact": vak_ooxml::diff::impact(current.as_ref(), &draft),
968. 968 });
969. 969 let choices = match lineage {
970. 970 None => Err("the draft was not made by office_apply in this conversation".to_string()),
971. 971 Some(lineage) => lineage_source(lineage, target_of(after)).and_then(|source| {
972. 972 vak_ooxml::review::choices(
973. 973 &source,
974. 974 &lineage.ops,
975. 975 &lineage_context(lineage),
976. 976 vak_ooxml::Limits::default(),
977. 977 target_of(after),
978. 978 &draft,
979. 979 )
980. 980 }),
981. 981 };
982. 982 match choices {
983. 983 Ok(choices) => body["choices"] = serde_json::json!(choices),
984. 984 Err(reason) => body["choices_unavailable"] = Value::String(reason),
985. 985 }
986. 986 serde_json::to_string(&body).map_err(|error| error.to_string())
987. 987}
988. 988
989. 989fn office_narrow_in_worker(
990. 990 lineage: &OfficeLineage,
991. 991 draft: &Path,
992. 992 keep: &[String],
993. 993 out: &Path,
994. 994) -> Result<String, String> {
995. 995 let source = lineage_source(lineage, target_of(out))?;
996. 996 let draft = read_document(draft)?;
997. 997 let applied = vak_ooxml::review::narrow(
998. 998 &source,
999. 999 &lineage.ops,
1000. 1000 keep,
1001. 1001 &lineage_context(lineage),
1002. 1002 vak_ooxml::Limits::default(),
1003. 1003 target_of(out),
1004. 1004 &draft,
1005. 1005 )
1006. 1006 .map_err(|error| format!("nothing was written: {error}"))?;
1007. 1007 crate::office_apply::write_atomically(out, &applied.bytes)?;
1008. 1008 serde_json::to_string(&serde_json::json!({
1009. 1009 "results": applied.results,
1010. 1010 "sha256": crate::office_apply::sha256_hex(&applied.bytes),
1011. 1011 }))
1012. 1012 .map_err(|error| error.to_string())
1013. 1013}
1014. 1014
1015. 1015/// Spawns one worker for `task` under a network-denied sandbox that can
1016. 1016/// read `roots` and write nothing, or only `roots[0]` when `writes_first`,
1017. 1017/// and returns its answer's content. Every failure, including the deadline,
1018. 1018/// is an error; nothing is guessed.
1019. 1019async fn run_task(
1020. 1020 worker_exe: &Path,
1021. 1021 cwd: &Path,
1022. 1022 roots: &[&Path],
1023. 1023 writes_first: bool,
1024. 1024 task: WorkerTask,
1025. 1025) -> Result<String, String> {
1026. 1026 if !worker_exe.is_file() {
1027. 1027 return Err(format!(
1028. 1028 "worker executable not found: {}",
1029. 1029 worker_exe.display()
1030. 1030 ));
1031. 1031 }
1032. 1032 let request = WorkerRequest {
1033. 1033 version: PROTOCOL_VERSION,
1034. 1034 task,
1035. 1035 };
1036. 1036 let payload =
1037. 1037 serde_json::to_vec(&request).map_err(|error| format!("request encode failed: {error}"))?;
1038. 1038 let worker_command = format!(
1039. 1039 "{} {}",
1040. 1040 shell_quote(&worker_exe.display().to_string()),
1041. 1041 WORKER_SUBCOMMAND
1042. 1042 );
1043. 1043 let effective = match verification_sandbox(roots, writes_first, worker_exe) {
1044. 1044 Some(sandbox) => sandbox.wrap(&worker_command),
1045. 1045 None => worker_command,
1046. 1046 };
1047. 1047 let mut command = tokio::process::Command::new(crate::bash::POSIX_SHELL);
1048. 1048 command
1049. 1049 .arg("-c")
1050. 1050 .arg(effective)
1051. 1051 .current_dir(cwd)
1052. 1052 .stdin(Stdio::piped())
1053. 1053 .stdout(Stdio::piped())
1054. 1054 .stderr(Stdio::piped())
1055. 1055 .kill_on_drop(true);
1056. 1056 crate::bash::scrub_environment(&mut command);
1057. 1057 command.env(WORKER_ENV, "1");
1058. 1058 crate::bash::isolate_process_group(&mut command);
1059. 1059 let mut child = command
1060. 1060 .spawn()
1061. 1061 .map_err(|error| format!("worker spawn failed: {error}"))?;
1062. 1062 let pid = child.id();
1063. 1063 let Some(mut stdin) = child.stdin.take() else {
1064. 1064 crate::bash::kill_process_group(&pid);
1065. 1065 return Err("worker has no stdin".into());
1066. 1066 };
1067. 1067 if let Err(error) = stdin.write_all(&payload).await {
1068. 1068 crate::bash::kill_process_group(&pid);
1069. 1069 return Err(format!("worker request failed: {error}"));
1070. 1070 }
1071. 1071 drop(stdin);
1072. 1072 let (Some(stdout), Some(stderr)) = (child.stdout.take(), child.stderr.take()) else {
1073. 1073 crate::bash::kill_process_group(&pid);
1074. 1074 return Err("worker has no output pipes".into());
1075. 1075 };
1076. 1076 let stdout_reader = tokio::spawn(async move {
1077. 1077 let mut bytes = Vec::new();
1078. 1078 let _ = stdout
1079. 1079 .take(MAX_PROTOCOL_BYTES + 1)
1080. 1080 .read_to_end(&mut bytes)
1081. 1081 .await;
1082. 1082 bytes
1083. 1083 });
1084. 1084 let stderr_reader = tokio::spawn(async move {
1085. 1085 let mut bytes = Vec::new();
1086. 1086 let _ = stderr.take(64 * 1024).read_to_end(&mut bytes).await;
1087. 1087 bytes
1088. 1088 });
1089. 1089 let status = match tokio::time::timeout(VERIFY_DEADLINE, child.wait()).await {
1090. 1090 Ok(Ok(status)) => status,
1091. 1091 Ok(Err(error)) => return Err(format!("worker wait failed: {error}")),
1092. 1092 Err(_) => {
1093. 1093 crate::bash::kill_process_group(&pid);
1094. 1094 let _ = child.wait().await;
1095. 1095 return Err(format!(
1096. 1096 "worker exceeded the {}s deadline",
1097. 1097 VERIFY_DEADLINE.as_secs()
1098. 1098 ));
1099. 1099 }
1100. 1100 };
1101. 1101 let stdout = stdout_reader.await.unwrap_or_default();
1102. 1102 let stderr = stderr_reader.await.unwrap_or_default();
1103. 1103 if !status.success() {
1104. 1104 return Err(format!(
1105. 1105 "worker exited with {}: {}",
1106. 1106 status.code().unwrap_or(-1),
1107. 1107 String::from_utf8_lossy(&stderr).trim()
1108. 1108 ));
1109. 1109 }
1110. 1110 if stdout.len() as u64 > MAX_PROTOCOL_BYTES {
1111. 1111 return Err("worker output exceeded the protocol limit".into());
1112. 1112 }
1113. 1113 let response: WorkerResponse = serde_json::from_slice(&stdout)
1114. 1114 .map_err(|error| format!("worker returned invalid protocol: {error}"))?;
1115. 1115 if response.version != PROTOCOL_VERSION || response.is_error {
1116. 1116 return Err(format!("worker refused the request: {}", response.content));
1117. 1117 }
1118. 1118 Ok(response.content)
1119. 1119}
1120. 1120
1121. 1121/// Network-denied, able to read `roots` and the worker executable, and to
1122. 1122/// write `roots[0]` only when `writes_first`. `None` where no OS backend
1123. 1123/// exists; the in-code bounds of each reader then stand alone.
1124. 1124fn verification_sandbox(
1125. 1125 roots: &[&Path],
1126. 1126 writes_first: bool,
1127. 1127 worker_exe: &Path,
1128. 1128) -> Option<Arc<dyn crate::sandbox::Sandbox>> {
1129. 1129 let (first, rest) = roots.split_first()?;
1130. 1130 let mut extra: Vec<PathBuf> = rest
1131. 1131 .iter()
1132. 1132 .filter_map(|root| root.canonicalize().ok())
1133. 1133 .collect();
1134. 1134 extra.extend(worker_exe.parent().map(Path::to_path_buf));
1135. 1135 let mode = if writes_first {
1136. 1136 crate::sandbox::SandboxMode::WorkspaceWrite
1137. 1137 } else {
1138. 1138 crate::sandbox::SandboxMode::ReadOnly
1139. 1139 };
1140. 1140 #[cfg(target_os = "macos")]
1141. 1141 {
1142. 1142 let mut sandbox = crate::sandbox::Seatbelt::task_copy(mode, first);
1143. 1143 sandbox.read_paths.extend(extra);
1144. 1144 Some(Arc::new(sandbox))
1145. 1145 }
1146. 1146 #[cfg(target_os = "linux")]
1147. 1147 {
1148. 1148 let mut sandbox = crate::landlock::Landlock::task_copy(mode, first);
1149. 1149 sandbox.read_paths.extend(extra);
1150. 1150 Some(Arc::new(sandbox))
1151. 1151 }
1152. 1152 #[cfg(not(any(target_os = "macos", target_os = "linux")))]
1153. 1153 {
1154. 1154 let _ = (first, extra, mode);
1155. 1155 None
1156. 1156 }
1157. 1157}
1158. 1158
1159. 1159fn shell_quote(value: &str) -> String {
1160. 1160 let mut out = String::with_capacity(value.len() + 2);
1161. 1161 out.push('\'');
1162. 1162 for character in value.chars() {
1163. 1163 if character == '\'' {
1164. 1164 out.push_str("'\\''");
1165. 1165 } else {
1166. 1166 out.push(character);
1167. 1167 }
1168. 1168 }
1169. 1169 out.push('\'');
1170. 1170 out
1171. 1171}
1172. 1172
1173. 1173/// Decides how a sandbox wrapper applies to a brokered tool invocation.
1174. 1174///
1175. 1175/// Two strategies:
1176. 1176/// - **`WorkerProcess` target** (Seatbelt/Landlock): the entire worker
1177. 1177/// process is wrapped, so the sandbox binary itself is sandboxed at exec.
1178. 1178/// - **`ToolCommand` target** (Docker): the worker runs on the host and only
1179. 1179/// the tool's own command is wrapped, so the model-controlled shell lands
1180. 1180/// inside the container.
1181. 1181///
1182. 1182/// Returns the effective shell command string and mutates `request_args`
1183. 1183/// in place when the command-level wrapping path is taken.
1184. 1184fn resolve_worker_command(
1185. 1185 sandbox: Option<&dyn crate::sandbox::Sandbox>,
1186. 1186 tool: &str,
1187. 1187 request_args: &mut serde_json::Value,
1188. 1188 worker_command: &str,
1189. 1189) -> String {
1190. 1190 match sandbox {
1191. 1191 Some(sb) if sb.target() == SandboxTarget::WorkerProcess => sb.wrap(worker_command),
1192. 1192 Some(sb) => {
1193. 1193 if tool == "bash"
1194. 1194 && let Some(command) = request_args
1195. 1195 .get("command")
1196. 1196 .and_then(serde_json::Value::as_str)
1197. 1197 {
1198. 1198 request_args["command"] = serde_json::Value::String(sb.wrap(command));
1199. 1199 }
1200. 1200 worker_command.to_string()
1201. 1201 }
1202. 1202 None => worker_command.to_string(),
1203. 1203 }
1204. 1204}
1205. 1205
1206. 1206#[cfg(test)]
1207. 1207mod tests {
1208. 1208 #![allow(clippy::unwrap_used, clippy::expect_used)]
1209. 1209
1210. 1210 use super::*;
1211. 1211 use crate::sandbox::{Sandbox, SandboxMode, Seatbelt};
1212. 1212 use serde_json::json;
1213. 1213
1214. 1214 #[test]
1215. 1215 fn no_sandbox_passes_worker_command_through() {
1216. 1216 let worker_cmd = "/path/to/vak __tool_worker";
1217. 1217 let mut args = json!({"command": "echo hi"});
1218. 1218 let effective = resolve_worker_command(None, "bash", &mut args, worker_cmd);
1219. 1219 assert_eq!(effective, worker_cmd);
1220. 1220 // args untouched
1221. 1221 assert_eq!(args["command"], json!("echo hi"));
1222. 1222 }
1223. 1223
1224. 1224 #[test]
1225. 1225 fn worker_process_target_wraps_the_worker_executable() {
1226. 1226 let dir = std::env::temp_dir();
1227. 1227 let sb: Arc<dyn Sandbox> = Arc::new(Seatbelt::new(SandboxMode::ReadOnly, &dir));
1228. 1228 assert_eq!(sb.target(), SandboxTarget::WorkerProcess);
1229. 1229 let worker_cmd = "/vak __tool_worker";
1230. 1230 let mut args = json!({"command": "echo hi"});
1231. 1231 let effective = resolve_worker_command(Some(&*sb), "bash", &mut args, worker_cmd);
1232. 1232 // The worker executable itself — not the inner bash command — is
1233. 1233 // wrapped, so the whole broker runs under sandbox-exec.
1234. 1234 assert!(
1235. 1235 effective.starts_with("sandbox-exec -p "),
1236. 1236 "expected sandbox-exec wrapper, got: {effective}"
1237. 1237 );
1238. 1238 assert!(effective.contains("__tool_worker"));
1239. 1239 // The inner bash command must NOT be wrapped — it travels through
1240. 1240 // the worker protocol, not the shell wrapper.
1241. 1241 assert!(!effective.contains("echo hi"));
1242. 1242 }
1243. 1243
1244. 1244 #[test]
1245. 1245 fn tool_command_target_wraps_bash_command_in_args() {
1246. 1246 // A Docker-style sandbox (ToolCommand target) must wrap the bash
1247. 1247 // command inside the worker request args — the host worker stays
1248. 1248 // a protocol adapter and only the model-controlled shell is sandboxed.
1249. 1249 let sb: Arc<dyn Sandbox> = Arc::new(DockerStub {
1250. 1250 target: SandboxTarget::ToolCommand,
1251. 1251 wrap_fn: |cmd| format!("docker-run-wrapped({})", cmd),
1252. 1252 });
1253. 1253 assert_eq!(sb.target(), SandboxTarget::ToolCommand);
1254. 1254
1255. 1255 let worker_cmd = "/vak __tool_worker";
1256. 1256
1257. 1257 // bash tool: command gets wrapped in the args
1258. 1258 let mut args = json!({"command": "echo from-bash"});
1259. 1259 let effective = resolve_worker_command(Some(&*sb), "bash", &mut args, worker_cmd);
1260. 1260 assert_eq!(
1261. 1261 effective, worker_cmd,
1262. 1262 "worker command passes through unwrapped"
1263. 1263 );
1264. 1264 assert_eq!(
1265. 1265 args["command"],
1266. 1266 json!("docker-run-wrapped(echo from-bash)"),
1267. 1267 "bash command must be wrapped in the request args"
1268. 1268 );
1269. 1269
1270. 1270 // non-bash tool: args untouched, worker command passes through
1271. 1271 let mut args2 = json!({"path": "src/main.rs"});
1272. 1272 let effective2 = resolve_worker_command(Some(&*sb), "read", &mut args2, worker_cmd);
1273. 1273 assert_eq!(effective2, worker_cmd);
1274. 1274 assert_eq!(args2["path"], json!("src/main.rs"));
1275. 1275 }
1276. 1276
1277. 1277 #[test]
1278. 1278 fn tool_command_target_does_not_wrap_bash_without_command_arg() {
1279. 1279 let sb: Arc<dyn Sandbox> = Arc::new(DockerStub {
1280. 1280 target: SandboxTarget::ToolCommand,
1281. 1281 wrap_fn: |cmd| format!("wrapped({})", cmd),
1282. 1282 });
1283. 1283 let worker_cmd = "/vak __tool_worker";
1284. 1284 let mut args = json!({});
1285. 1285 let effective = resolve_worker_command(Some(&*sb), "bash", &mut args, worker_cmd);
1286. 1286 assert_eq!(effective, worker_cmd);
1287. 1287 assert_eq!(
1288. 1288 args["command"],
1289. 1289 json!(Value::Null),
1290. 1290 "no command key to wrap"
1291. 1291 );
1292. 1292 }
1293. 1293
1294. 1294 /// Minimal Sandbox stub that lets us test ToolCommand-target wrapping
1295. 1295 /// without a Docker daemon.
1296. 1296 struct DockerStub {
1297. 1297 target: SandboxTarget,
1298. 1298 wrap_fn: fn(&str) -> String,
1299. 1299 }
1300. 1300
1301. 1301 impl Sandbox for DockerStub {
1302. 1302 fn name(&self) -> &str {
1303. 1303 "docker-stub"
1304. 1304 }
1305. 1305 fn wrap(&self, command: &str) -> String {
1306. 1306 (self.wrap_fn)(command)
1307. 1307 }
1308. 1308 fn target(&self) -> SandboxTarget {
1309. 1309 self.target
1310. 1310 }
1311. 1311 }
1312. 1312}
1313. 1313